From d4952f4d7b12d09b7c3fd54949fb00ea14f2bee8 Mon Sep 17 00:00:00 2001 From: datron Date: Thu, 16 Jul 2026 16:17:35 +0530 Subject: [PATCH 01/12] feat: add background job manager database models and request/response types Signed-off-by: datron --- .../2026-07-16-000001_job_manager/down.sql | 20 + .../2026-07-16-000001_job_manager/up.sql | 52 ++ crates/superposition_types/src/api.rs | 1 + crates/superposition_types/src/api/jobs.rs | 114 ++++ .../src/database/models.rs | 148 +++++ .../src/database/models/others.rs | 23 +- .../src/database/schema.rs | 592 +----------------- .../src/database/superposition_schema.rs | 29 +- superposition.sql | 53 ++ workspace_template.sql | 16 +- 10 files changed, 470 insertions(+), 578 deletions(-) create mode 100644 crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql create mode 100644 crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql create mode 100644 crates/superposition_types/src/api/jobs.rs diff --git a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql new file mode 100644 index 000000000..c65b7e635 --- /dev/null +++ b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql @@ -0,0 +1,20 @@ +DROP TABLE IF EXISTS superposition.job_manager; + +DROP INDEX IF EXISTS superposition.idx_job_manager_workspace_schema; +DROP INDEX IF EXISTS superposition.idx_job_manager_status; +DROP INDEX IF EXISTS superposition.idx_job_manager_type; +DROP INDEX IF EXISTS superposition.idx_job_manager_kronos_job_id; +DROP INDEX IF EXISTS superposition.idx_job_manager_status_job_type; +DROP INDEX IF EXISTS superposition.idx_job_manager_created_at; + +DO $$ BEGIN + DROP TYPE public.background_job_type; +EXCEPTION + WHEN undefined_object THEN null; +END $$; + +DO $$ BEGIN + DROP TYPE public.background_job_status; +EXCEPTION + WHEN undefined_object THEN null; +END $$; diff --git a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql new file mode 100644 index 000000000..f641f27bb --- /dev/null +++ b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql @@ -0,0 +1,52 @@ +DO $$ BEGIN + CREATE TYPE public.background_job_type AS ENUM ( + 'WEBHOOK', + 'PRIORITY_RECOMPUTE', + 'REDUCE' + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +DO $$ BEGIN + CREATE TYPE public.background_job_status AS ENUM ( + 'CREATED', + 'SCHEDULED', + 'INPROGRESS', + 'FAILED', + 'COMPLETED' + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE TABLE IF NOT EXISTS superposition.job_manager ( + id BIGINT PRIMARY KEY, + kronos_job_id TEXT NOT NULL, + description TEXT NOT NULL, + job_type public.background_job_type NOT NULL, + status public.background_job_status NOT NULL, + name TEXT NOT NULL, + progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100), + workspace_schema TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + logs TEXT NOT NULL DEFAULT '' +); + +CREATE INDEX IF NOT EXISTS idx_job_manager_workspace_schema + ON superposition.job_manager (workspace_schema); + +CREATE INDEX IF NOT EXISTS idx_job_manager_kronos_job_id + ON superposition.job_manager (kronos_job_id); + +CREATE INDEX IF NOT EXISTS idx_job_manager_type + ON superposition.job_manager (job_type); + +CREATE INDEX IF NOT EXISTS idx_job_manager_status + ON superposition.job_manager (status); + +CREATE INDEX IF NOT EXISTS idx_job_manager_status_job_type + ON superposition.job_manager (status, job_type); + +CREATE INDEX IF NOT EXISTS idx_job_manager_created_at + ON superposition.job_manager (created_at DESC); \ No newline at end of file diff --git a/crates/superposition_types/src/api.rs b/crates/superposition_types/src/api.rs index 0e084ba86..084e9e6d0 100644 --- a/crates/superposition_types/src/api.rs +++ b/crates/superposition_types/src/api.rs @@ -11,6 +11,7 @@ pub mod experiment_groups; #[cfg(feature = "experimentation")] pub mod experiments; pub mod functions; +pub mod jobs; pub mod organisation; pub mod secrets; pub mod type_templates; diff --git a/crates/superposition_types/src/api/jobs.rs b/crates/superposition_types/src/api/jobs.rs new file mode 100644 index 000000000..0f09f706b --- /dev/null +++ b/crates/superposition_types/src/api/jobs.rs @@ -0,0 +1,114 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::database::models::{ + BackgroundJob, BackgroundJobStatus, BackgroundJobType, +}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "job_type", content = "job_data")] +pub enum JobRequest { + Webhook(DispatchWebhookRequest), + PriorityRecompute(PriorityRecomputeRequest), + Reduce(ReduceRequest), +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DispatchWebhookRequest { + pub webhook_name: String, + pub data: Value, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct PriorityRecomputeRequest {} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ReduceRequest { + #[serde(default)] + pub approve: bool, +} + +impl JobRequest { + pub fn job_type(&self) -> BackgroundJobType { + match self { + Self::Webhook(_) => BackgroundJobType::Webhook, + Self::PriorityRecompute(_) => BackgroundJobType::PriorityRecompute, + Self::Reduce(_) => BackgroundJobType::Reduce, + } + } + + pub fn job_name(&self) -> String { + match self { + Self::Webhook(r) => r.webhook_name.clone(), + Self::PriorityRecompute(_) => "priority_recompute".to_string(), + Self::Reduce(_) => "reduce".to_string(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobResponse { + #[serde(with = "crate::database::models::i64_formatter")] + pub id: i64, + pub kronos_job_id: String, + pub description: String, + #[serde(rename = "type")] + pub job_type: BackgroundJobType, + pub status: BackgroundJobStatus, + pub name: String, + pub progress: i32, + pub workspace_schema: String, + pub created_at: DateTime, + pub logs: String, +} + +impl From for JobResponse { + fn from(job: BackgroundJob) -> Self { + Self { + id: job.id, + kronos_job_id: job.kronos_job_id, + description: job.description, + job_type: job.job_type, + status: job.status, + name: job.name, + progress: job.progress, + workspace_schema: job.workspace_schema, + created_at: job.created_at, + logs: job.logs, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobCreateResponse { + #[serde(with = "crate::database::models::i64_formatter")] + pub id: i64, + pub kronos_job_id: String, + pub status: BackgroundJobStatus, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ExecutionDetails { + pub attempt_count: Option, + pub max_attempts: Option, + pub started_at: Option>, + pub completed_at: Option>, + pub duration_ms: Option, + pub execution_status: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobDetailResponse { + #[serde(flatten)] + pub job: JobResponse, + pub execution: Option, +} + +#[derive(Debug, Deserialize, Serialize, Default, Clone)] +pub struct JobListFilters { + #[serde(default)] + pub status: Option, + #[serde(default)] + pub job_type: Option, +} diff --git a/crates/superposition_types/src/database/models.rs b/crates/superposition_types/src/database/models.rs index c6544ba24..a013195b4 100644 --- a/crates/superposition_types/src/database/models.rs +++ b/crates/superposition_types/src/database/models.rs @@ -288,6 +288,154 @@ pub struct Workspace { pub workspace_lock_expires_at: Option>, } +#[derive( + Debug, Clone, Copy, PartialEq, Deserialize, Serialize, strum_macros::Display, +)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +#[strum(serialize_all = "SCREAMING_SNAKE_CASE")] +#[cfg_attr( + feature = "diesel_derives", + derive(diesel_derive_enum::DbEnum, QueryId) +)] +#[cfg_attr(feature = "diesel_derives", DbValueStyle = "SCREAMING_SNAKE_CASE")] +#[cfg_attr( + feature = "diesel_derives", + ExistingTypePath = "crate::database::superposition_schema::superposition::sql_types::BackgroundJobType" +)] +pub enum BackgroundJobType { + Webhook, + PriorityRecompute, + Reduce, +} + +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Deserialize, + Serialize, + strum_macros::Display, + strum_macros::EnumIter, +)] +#[serde(rename_all = "UPPERCASE")] +#[strum(serialize_all = "UPPERCASE")] +#[cfg_attr( + feature = "diesel_derives", + derive(diesel_derive_enum::DbEnum, QueryId) +)] +#[cfg_attr(feature = "diesel_derives", DbValueStyle = "UPPERCASE")] +#[cfg_attr( + feature = "diesel_derives", + ExistingTypePath = "crate::database::superposition_schema::superposition::sql_types::BackgroundJobStatus" +)] +pub enum BackgroundJobStatus { + Created, + Scheduled, + Inprogress, + Failed, + Completed, +} + +#[derive(Clone, Debug, PartialEq)] +#[cfg_attr( + all( + feature = "diesel_derives", + not(feature = "disable_db_data_validation") + ), + derive(TextFromSql) +)] +#[cfg_attr( + all(feature = "diesel_derives", feature = "disable_db_data_validation"), + derive(TextFromSqlNoValidation) +)] +#[cfg_attr( + feature = "diesel_derives", + derive(AsExpression, FromSqlRow, TextToSql) +)] +#[cfg_attr(feature = "diesel_derives", diesel(sql_type = Text))] +pub enum JobWorkspace { + Global, + Workspace(String), +} + +impl Serialize for JobWorkspace { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + match self { + Self::Global => serializer.serialize_str("GLOBAL"), + Self::Workspace(schema) => serializer.serialize_str(schema), + } + } +} + +impl JobWorkspace { + pub const GLOBAL: &'static str = "global"; + + pub fn as_db_string(&self) -> String { + match self { + Self::Global => Self::GLOBAL.to_string(), + Self::Workspace(schema) => schema.clone(), + } + } + + pub fn from_workspace_schema(schema: &str) -> Self { + Self::Workspace(schema.to_string()) + } +} + +impl TryFrom for JobWorkspace { + type Error = String; + + fn try_from(value: String) -> Result { + if value == Self::GLOBAL { + Ok(Self::Global) + } else { + Ok(Self::Workspace(value)) + } + } +} + +impl From<&JobWorkspace> for String { + fn from(value: &JobWorkspace) -> Self { + value.as_db_string() + } +} + +#[cfg(feature = "disable_db_data_validation")] +impl DisableDBValidation for JobWorkspace { + type Source = String; + fn from_db_unvalidated(data: Self::Source) -> Self { + Self::try_from(data).unwrap_or(Self::Global) + } +} + +#[derive(Clone, Serialize, Debug)] +#[cfg_attr( + feature = "diesel_derives", + derive(Queryable, Selectable, Insertable, AsChangeset) +)] +#[cfg_attr(feature = "diesel_derives", diesel(check_for_backend(diesel::pg::Pg)))] +#[cfg_attr(feature = "diesel_derives", diesel(primary_key(id)))] +#[cfg_attr( + feature = "diesel_derives", + diesel(table_name = job_manager) +)] +pub struct BackgroundJob { + pub id: i64, + pub kronos_job_id: String, + pub description: String, + pub job_type: BackgroundJobType, + pub status: BackgroundJobStatus, + pub name: String, + pub progress: i32, + pub workspace_schema: JobWorkspace, + pub created_at: DateTime, + pub logs: String, +} + #[derive(Clone, Serialize, Deserialize, Debug)] #[serde(rename_all = "lowercase")] pub enum MetricSource { diff --git a/crates/superposition_types/src/database/models/others.rs b/crates/superposition_types/src/database/models/others.rs index df412789e..f0486c1a3 100644 --- a/crates/superposition_types/src/database/models/others.rs +++ b/crates/superposition_types/src/database/models/others.rs @@ -19,8 +19,10 @@ use superposition_derives::{TextFromSql, TextToSql}; use crate::RegexEnum; #[cfg(feature = "diesel_derives")] -use super::super::schema::{secrets, variables, webhooks}; -use super::{ChangeReason, Description, NonEmptyString}; +use super::super::schema::{job_manager, secrets, variables, webhooks}; +use super::{ + BackgroundJobStatus, BackgroundJobType, ChangeReason, Description, NonEmptyString, +}; #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Default)] #[cfg_attr( @@ -270,3 +272,20 @@ pub struct Secret { pub created_by: String, pub last_modified_by: String, } + +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] +#[cfg_attr(feature = "diesel_derives", derive(Queryable, Selectable))] +#[cfg_attr(feature = "diesel_derives", diesel(check_for_backend(diesel::pg::Pg)))] +#[cfg_attr(feature = "diesel_derives", diesel(table_name = job_manager))] +#[cfg_attr(feature = "diesel_derives", diesel(primary_key(id)))] +pub struct WorkspaceJobView { + pub id: i64, + pub kronos_job_id: String, + pub description: String, + pub job_type: BackgroundJobType, + pub status: BackgroundJobStatus, + pub name: String, + pub progress: i32, + pub created_at: DateTime, + pub logs: String, +} diff --git a/crates/superposition_types/src/database/schema.rs b/crates/superposition_types/src/database/schema.rs index d2747fe4c..e3036bd69 100644 --- a/crates/superposition_types/src/database/schema.rs +++ b/crates/superposition_types/src/database/schema.rs @@ -100,539 +100,6 @@ diesel::table! { } } -diesel::table! { - event_log_y2023m08 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2023m09 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2023m10 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2023m11 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2023m12 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m01 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m02 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m03 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m04 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m05 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m06 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m07 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m08 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m09 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m10 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m11 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2024m12 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m01 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m02 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m03 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m04 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m05 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m06 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m07 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m08 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m09 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m10 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m11 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2025m12 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m01 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m02 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m03 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m04 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m05 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m06 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m07 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m08 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m09 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m10 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m11 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - -diesel::table! { - event_log_y2026m12 (id, timestamp) { - id -> Uuid, - table_name -> Text, - user_name -> Text, - timestamp -> Timestamptz, - action -> Text, - original_data -> Nullable, - new_data -> Nullable, - query -> Text, - } -} - diesel::table! { use diesel::sql_types::*; use super::sql_types::{ExperimentStatusType, ExperimentType}; @@ -776,6 +243,22 @@ diesel::table! { } } +diesel::table! { + use diesel::sql_types::*; + use super::super::superposition_schema::superposition::sql_types::{BackgroundJobType, BackgroundJobStatus}; + job_manager (id) { + id -> Int8, + kronos_job_id -> Text, + description -> Text, + job_type -> BackgroundJobType, + status -> BackgroundJobStatus, + name -> Text, + progress -> Int4, + created_at -> Timestamptz, + logs -> Text, + } +} + diesel::joinable!(default_configs -> functions (value_validation_function_name)); diesel::joinable!(dimensions -> functions (value_validation_function_name)); @@ -785,51 +268,12 @@ diesel::allow_tables_to_appear_in_same_query!( default_configs, dimensions, event_log, - event_log_y2023m08, - event_log_y2023m09, - event_log_y2023m10, - event_log_y2023m11, - event_log_y2023m12, - event_log_y2024m01, - event_log_y2024m02, - event_log_y2024m03, - event_log_y2024m04, - event_log_y2024m05, - event_log_y2024m06, - event_log_y2024m07, - event_log_y2024m08, - event_log_y2024m09, - event_log_y2024m10, - event_log_y2024m11, - event_log_y2024m12, - event_log_y2025m01, - event_log_y2025m02, - event_log_y2025m03, - event_log_y2025m04, - event_log_y2025m05, - event_log_y2025m06, - event_log_y2025m07, - event_log_y2025m08, - event_log_y2025m09, - event_log_y2025m10, - event_log_y2025m11, - event_log_y2025m12, - event_log_y2026m01, - event_log_y2026m02, - event_log_y2026m03, - event_log_y2026m04, - event_log_y2026m05, - event_log_y2026m06, - event_log_y2026m07, - event_log_y2026m08, - event_log_y2026m09, - event_log_y2026m10, - event_log_y2026m11, - event_log_y2026m12, experiments, experiment_groups, functions, type_templates, webhooks, variables, + secrets, + job_manager, ); diff --git a/crates/superposition_types/src/database/superposition_schema.rs b/crates/superposition_types/src/database/superposition_schema.rs index 5eff9abe9..721fd5b0f 100644 --- a/crates/superposition_types/src/database/superposition_schema.rs +++ b/crates/superposition_types/src/database/superposition_schema.rs @@ -9,6 +9,14 @@ pub mod superposition { #[derive(diesel::sql_types::SqlType)] #[diesel(postgres_type(name = "workspace_status", schema = "superposition"))] pub struct WorkspaceStatus; + + #[derive(diesel::sql_types::SqlType)] + #[diesel(postgres_type(name = "background_job_type", schema = "public"))] + pub struct BackgroundJobType; + + #[derive(diesel::sql_types::SqlType)] + #[diesel(postgres_type(name = "background_job_status", schema = "public"))] + pub struct BackgroundJobStatus; } diesel::table! { @@ -70,7 +78,26 @@ pub mod superposition { } } + diesel::table! { + use diesel::sql_types::*; + use super::sql_types::BackgroundJobType; + use super::sql_types::BackgroundJobStatus; + + superposition.job_manager (id) { + id -> Int8, + kronos_job_id -> Text, + description -> Text, + job_type -> BackgroundJobType, + status -> BackgroundJobStatus, + name -> Text, + progress -> Int4, + workspace_schema -> Text, + created_at -> Timestamptz, + logs -> Text, + } + } + diesel::joinable!(workspaces -> organisations (organisation_id)); - diesel::allow_tables_to_appear_in_same_query!(organisations, workspaces,); + diesel::allow_tables_to_appear_in_same_query!(organisations, workspaces, job_manager,); } diff --git a/superposition.sql b/superposition.sql index 6c643229f..0a0685ad4 100644 --- a/superposition.sql +++ b/superposition.sql @@ -155,3 +155,56 @@ ADD COLUMN IF NOT EXISTS workspace_locked_by TEXT, ADD COLUMN IF NOT EXISTS workspace_lock_acquired_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS workspace_lock_expires_at TIMESTAMPTZ; +DO $$ BEGIN + CREATE TYPE public.background_job_type AS ENUM ( + 'WEBHOOK', + 'PRIORITY_RECOMPUTE', + 'REDUCE' + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +DO $$ BEGIN + CREATE TYPE public.background_job_status AS ENUM ( + 'CREATED', + 'SCHEDULED', + 'INPROGRESS', + 'FAILED', + 'COMPLETED' + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE TABLE IF NOT EXISTS superposition.job_manager ( + id BIGINT PRIMARY KEY, + kronos_job_id TEXT NOT NULL, + description TEXT NOT NULL, + job_type public.background_job_type NOT NULL, + status public.background_job_status NOT NULL, + name TEXT NOT NULL, + progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100), + workspace_schema TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + logs TEXT NOT NULL DEFAULT '' +); + +CREATE INDEX IF NOT EXISTS idx_job_manager_workspace_schema + ON superposition.job_manager (workspace_schema); + +CREATE INDEX IF NOT EXISTS idx_job_manager_kronos_job_id + ON superposition.job_manager (kronos_job_id); + +CREATE INDEX IF NOT EXISTS idx_job_manager_type + ON superposition.job_manager (job_type); + +CREATE INDEX IF NOT EXISTS idx_job_manager_status + ON superposition.job_manager (status); + +CREATE INDEX IF NOT EXISTS idx_job_manager_status_job_type + ON superposition.job_manager (status, job_type); + +CREATE INDEX IF NOT EXISTS idx_job_manager_created_at + ON superposition.job_manager (created_at DESC); + diff --git a/workspace_template.sql b/workspace_template.sql index c7942b7df..e868409af 100644 --- a/workspace_template.sql +++ b/workspace_template.sql @@ -1011,4 +1011,18 @@ SELECT 'base config version recording the addition of variantIds dimension' WHERE NOT EXISTS ( SELECT 1 FROM {replaceme}.config_versions -); \ No newline at end of file +); + +CREATE OR REPLACE VIEW {replaceme}.job_manager AS +SELECT + id, + kronos_job_id, + description, + job_type, + status, + name, + progress, + created_at, + logs +FROM superposition.job_manager +WHERE workspace_schema = '{replaceme}'; \ No newline at end of file From bfe92c0d01af300d85380b9b6e2ed20b70defbde Mon Sep 17 00:00:00 2001 From: datron Date: Fri, 17 Jul 2026 16:36:55 +0530 Subject: [PATCH 02/12] feat: add dispatch functions and API handlers Signed-off-by: datron --- crates/service_utils/src/kronos_dispatch.rs | 260 +++++++++++++++++- crates/superposition/src/app_state.rs | 6 +- crates/superposition/src/jobs.rs | 2 + crates/superposition/src/jobs/handlers.rs | 143 ++++++++++ crates/superposition/src/main.rs | 6 + .../superposition/src/workspace/handlers.rs | 14 +- crates/superposition_types/src/api/jobs.rs | 33 ++- .../src/database/models.rs | 11 +- crates/superposition_types/src/lib.rs | 2 + 9 files changed, 458 insertions(+), 19 deletions(-) create mode 100644 crates/superposition/src/jobs.rs create mode 100644 crates/superposition/src/jobs/handlers.rs diff --git a/crates/service_utils/src/kronos_dispatch.rs b/crates/service_utils/src/kronos_dispatch.rs index 966badf25..cf3af7af4 100644 --- a/crates/service_utils/src/kronos_dispatch.rs +++ b/crates/service_utils/src/kronos_dispatch.rs @@ -1,11 +1,28 @@ -use std::collections::HashMap; +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, +}; use base64::{Engine, engine::general_purpose}; +use chrono::Utc; +use diesel::{ExpressionMethods, QueryDsl, RunQueryDsl, SelectableHelper}; use kronos_common::{sqlx, tenant::SchemaProvider}; use kronos_worker::{JobTrigger, KronosClient}; use once_cell::sync::Lazy; use serde::Serialize; use serde_json::json; +use snowflake::SnowflakeIdGenerator; +use superposition_types::{ + DBConnection, + api::jobs::{JobCreateResponse, JobRequest}, + database::{ + models::{ + BackgroundJob, BackgroundJobStatus, JobWorkspace, others::WorkspaceJobView, + }, + schema::job_manager::dsl as job_manager_view_dsl, + superposition_schema::superposition::job_manager::dsl as job_manager_dsl, + }, +}; use crate::helpers::get_from_env_or_default; @@ -15,17 +32,10 @@ static CONFIG_REFERENCE_REGEX: Lazy = Lazy::new(|| { }); pub const DISPATCHER_ENDPOINT_NAME: &str = "superposition-webhook-dispatcher"; +pub const JOB_DISPATCHER_ENDPOINT_NAME: &str = "superposition-job-dispatcher"; pub const DISPATCHER_SECRET_NAME: &str = "superposition-internal-token"; -/// Username part of the Basic credential the dispatcher calls back with. pub const DISPATCHER_USERNAME: &str = "kronos-dispatcher"; -/// Build the Basic credential blob stored as the Kronos secret: -/// `base64("kronos-dispatcher:")`. Kronos stores it encrypted and sends -/// it verbatim in the Authorization header; SP decodes and verifies it. -pub fn dispatcher_basic_credential(dispatch_token: &str) -> String { - general_purpose::STANDARD.encode(format!("{DISPATCHER_USERNAME}:{dispatch_token}")) -} - pub struct SuperpositionSchemaProvider { pool: sqlx::PgPool, } @@ -104,7 +114,8 @@ pub async fn setup_dispatcher( } let dispatcher_url = format!("{superposition_host}/dispatch/webhook"); - let basic_credential = dispatcher_basic_credential(dispatch_token); + let basic_credential = general_purpose::STANDARD + .encode(format!("{DISPATCHER_USERNAME}:{dispatch_token}")); if let Err(e) = kronos_client .upsert_secret(workspace, DISPATCHER_SECRET_NAME, &basic_credential) .await @@ -125,6 +136,45 @@ pub async fn setup_dispatcher( } } +pub fn job_dispatcher_endpoint_spec(dispatcher_url: &str) -> serde_json::Value { + let timeout_ms: u64 = get_from_env_or_default("DISPATCHER_TIMEOUT_MS", 15000); + json!({ + "url": dispatcher_url, + "method": "POST", + "headers": { + "Authorization": format!("Basic {{{{secret.{DISPATCHER_SECRET_NAME}}}}}"), + "x-org-id": "{{input.org_id}}", + "x-workspace": "{{input.workspace}}" + }, + "timeout_ms": timeout_ms, + "expected_status_codes": [200] + }) +} + +/// Register the generic job dispatcher endpoint in Kronos. Called alongside `setup_dispatcher` +/// during app startup. Reuses the same secret as the webhook dispatcher. +pub async fn setup_job_dispatcher( + kronos_client: &dyn KronosClient, + workspace: &str, + superposition_host: &str, +) { + let dispatcher_url = format!("{superposition_host}/dispatch/job"); + if let Err(e) = kronos_client + .register_endpoint( + workspace, + JOB_DISPATCHER_ENDPOINT_NAME, + "HTTP", + job_dispatcher_endpoint_spec(&dispatcher_url), + Some(dispatcher_retry_policy()), + ) + .await + { + log::warn!( + "Kronos job dispatcher: endpoint register failed for '{workspace}': {e}" + ); + } +} + pub fn has_pattern_in_headers( headers: &serde_json::Map, ) -> (bool, bool) { @@ -197,3 +247,193 @@ pub async fn submit_webhook_job( ) .await } + +/// Submit a background job to Kronos and track it in the BJM table. +/// +/// 1. Generates a snowflake ID and inserts a `CREATED` entry into `superposition.job_manager`. +/// 2. Calls `kronos_client.create_job()` with the `JobRequest` as input. +/// 3. On success → updates BJM status to `SCHEDULED` and stores the Kronos job ID. +/// 4. On failure → updates BJM status to `FAILED` with the error in logs. +/// +#[allow(clippy::too_many_arguments)] +pub async fn submit_job( + kronos_client: &dyn KronosClient, + target_workspace: &str, + workspace: &JobWorkspace, + org_id: &str, + workspace_id: &str, + job_request: &JobRequest, + snowflake_generator: &Arc>, + conn: &mut DBConnection, + max_attempts: i64, + description: &str, +) -> anyhow::Result { + let job_id = { + let mut id_gen = snowflake_generator + .lock() + .map_err(|e| anyhow::anyhow!("snowflake lock failed: {e}"))?; + id_gen.real_time_generate() + }; + + let job_type = job_request.job_type(); + let job_name = job_request.job_name(); + let schema_str = workspace.as_db_string(); + + let bjm_entry = BackgroundJob { + id: job_id, + kronos_job_id: String::new(), + description: description.to_string(), + job_type, + status: BackgroundJobStatus::Created, + name: job_name.clone(), + progress: 0, + workspace_schema: workspace.clone(), + created_at: Utc::now(), + logs: String::new(), + }; + + diesel::insert_into(job_manager_dsl::job_manager) + .values(&bjm_entry) + .execute(conn)?; + + let job_request_value = serde_json::to_value(job_request)?; + let mut input = job_request_value; + if let Some(obj) = input.as_object_mut() { + obj.insert("org_id".to_string(), json!(org_id)); + obj.insert("workspace".to_string(), json!(workspace_id)); + obj.insert("job_id".to_string(), json!(job_id.to_string())); + } + + let idempotency_key = format!( + "{}_{}_{}_{}", + schema_str, + job_name, + job_type, + Utc::now().timestamp_millis() + ); + + match kronos_client + .create_job( + target_workspace, + JOB_DISPATCHER_ENDPOINT_NAME, + input, + max_attempts, + JobTrigger::Immediate, + Some(&idempotency_key), + ) + .await + { + Ok(kronos_job_id) => { + diesel::update( + job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id)), + ) + .set(( + job_manager_dsl::kronos_job_id.eq(&kronos_job_id), + job_manager_dsl::status.eq(BackgroundJobStatus::Scheduled), + )) + .execute(conn)?; + + Ok(JobCreateResponse { + id: job_id, + kronos_job_id, + status: BackgroundJobStatus::Scheduled, + }) + } + Err(e) => { + let error_msg = format!("Kronos job creation failed: {e}"); + log::error!("submit_job: {error_msg}"); + diesel::update( + job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id)), + ) + .set(( + job_manager_dsl::status.eq(BackgroundJobStatus::Failed), + job_manager_dsl::logs.eq(&error_msg), + )) + .execute(conn)?; + Err(e) + } + } +} + +pub fn get_job_by_id( + conn: &mut DBConnection, + workspace: &JobWorkspace, + job_id: i64, +) -> anyhow::Result { + let schema = workspace.as_db_string(); + job_manager_view_dsl::job_manager + .filter(job_manager_view_dsl::id.eq(job_id)) + .schema_name(&schema) + .select(WorkspaceJobView::as_select()) + .first::(conn) + .map_err(|e| anyhow::anyhow!("Failed to fetch job {job_id}: {e}")) +} + +pub fn list_jobs( + conn: &mut DBConnection, + workspace: &JobWorkspace, + job_type: Option, + status: Option, +) -> anyhow::Result> { + let schema = workspace.as_db_string(); + let mut query = job_manager_view_dsl::job_manager + .schema_name(&schema) + .select(WorkspaceJobView::as_select()) + .into_boxed(); + + if let Some(jt) = job_type { + query = query.filter(job_manager_view_dsl::job_type.eq(jt)); + } + if let Some(st) = status { + query = query.filter(job_manager_view_dsl::status.eq(st)); + } + + query + .order(job_manager_view_dsl::created_at.desc()) + .load::(conn) + .map_err(|e| anyhow::anyhow!("Failed to list jobs: {e}")) +} + +pub fn update_job_status( + conn: &mut DBConnection, + job_id: i64, + status: BackgroundJobStatus, +) -> anyhow::Result<()> { + diesel::update(job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id))) + .set(job_manager_dsl::status.eq(status)) + .execute(conn)?; + Ok(()) +} + +pub fn update_job_progress( + conn: &mut DBConnection, + job_id: i64, + progress: i32, +) -> anyhow::Result<()> { + diesel::update(job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id))) + .set(job_manager_dsl::progress.eq(progress)) + .execute(conn)?; + Ok(()) +} + +pub fn append_job_logs( + conn: &mut DBConnection, + job_id: i64, + log_line: &str, +) -> anyhow::Result<()> { + let current = job_manager_dsl::job_manager + .filter(job_manager_dsl::id.eq(job_id)) + .select(job_manager_dsl::logs) + .first::(conn)?; + + let new_logs = if current.is_empty() { + log_line.to_string() + } else { + format!("{current}\n{log_line}") + }; + + diesel::update(job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id))) + .set(job_manager_dsl::logs.eq(new_logs)) + .execute(conn)?; + Ok(()) +} diff --git a/crates/superposition/src/app_state.rs b/crates/superposition/src/app_state.rs index 072dc5332..19d5315cd 100644 --- a/crates/superposition/src/app_state.rs +++ b/crates/superposition/src/app_state.rs @@ -19,7 +19,9 @@ use service_utils::{ }, encryption::get_master_encryption_keys, helpers::{get_from_env_or_default, get_from_env_unsafe}, - kronos_dispatch::{SuperpositionSchemaProvider, setup_dispatcher}, + kronos_dispatch::{ + SuperpositionSchemaProvider, setup_dispatcher, setup_job_dispatcher, + }, service::types::{AppEnv, AppState, ExperimentationFlags}, }; use snowflake::SnowflakeIdGenerator; @@ -121,6 +123,8 @@ pub async fn get( ) .await; + setup_job_dispatcher(client.as_ref(), &workspace, &superposition_host).await; + (client, None, 0, Some(workspace)) } else { let database_url = get_database_url(kms_client, &app_env, None).await; diff --git a/crates/superposition/src/jobs.rs b/crates/superposition/src/jobs.rs new file mode 100644 index 000000000..e87b7d56c --- /dev/null +++ b/crates/superposition/src/jobs.rs @@ -0,0 +1,2 @@ +mod handlers; +pub use handlers::endpoints; diff --git a/crates/superposition/src/jobs/handlers.rs b/crates/superposition/src/jobs/handlers.rs new file mode 100644 index 000000000..c4c7e29e9 --- /dev/null +++ b/crates/superposition/src/jobs/handlers.rs @@ -0,0 +1,143 @@ +use actix_web::{ + HttpResponse, Scope, get, post, + web::{Data, Json, Path, Query}, +}; +use service_utils::{ + kronos_dispatch::{append_job_logs, get_job_by_id, list_jobs, update_job_status}, + service::types::{AppState, DbConnection, WorkspaceContext}, +}; +use superposition_derives::{authorized, declare_resource}; +use superposition_macros::unexpected_error; +use superposition_types::{ + api::jobs::{ExecutionDetails, JobDetailResponse, JobListFilters, JobResponse}, + database::models::{BackgroundJobStatus, JobWorkspace, others::WorkspaceJobView}, + result as superposition, +}; + +declare_resource!(WorkspaceJob); + +pub fn endpoints() -> Scope { + Scope::new("") + .service(list_handler) + .service(get_handler) + .service(cancel_handler) +} + +#[authorized] +#[get("")] +async fn list_handler( + workspace_context: WorkspaceContext, + db_conn: DbConnection, + filters: Query, +) -> superposition::Result>> { + let DbConnection(mut conn) = db_conn; + let filters = filters.into_inner(); + let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); + + let jobs = list_jobs(&mut conn, &job_workspace, filters.job_type, filters.status) + .map_err(|e| unexpected_error!("Failed to list jobs: {}", e))?; + + Ok(Json(jobs)) +} + +#[authorized] +#[get("/{job_id}")] +async fn get_handler( + workspace_context: WorkspaceContext, + state: Data, + db_conn: DbConnection, + job_id: Path, +) -> superposition::Result> { + let DbConnection(mut conn) = db_conn; + let job_id_str = job_id.into_inner(); + let job_id: i64 = job_id_str + .parse() + .map_err(|e| unexpected_error!("Invalid job_id '{}': {}", job_id_str, e))?; + + let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); + let job = get_job_by_id(&mut conn, &job_workspace, job_id) + .map_err(|e| unexpected_error!("Failed to fetch job: {}", e))?; + + let target_workspace = state + .kronos_workspace + .as_deref() + .unwrap_or(&workspace_context.schema_name); + + let execution = match state + .kronos_client + .get_execution(target_workspace, &job.kronos_job_id) + .await + { + Ok(Some(exec)) => Some(ExecutionDetails { + attempt_count: Some(exec.attempt_count), + max_attempts: Some(exec.max_attempts), + started_at: exec.started_at, + completed_at: exec.completed_at, + duration_ms: exec.duration_ms, + execution_status: Some(exec.status), + }), + Ok(None) => None, + Err(e) => { + log::warn!("Failed to fetch Kronos execution details: {}", e); + None + } + }; + let schema_name = workspace_context.schema_name.0; + Ok(Json(JobDetailResponse { + job: JobResponse::from_view(&job, &schema_name), + execution, + })) +} + +#[authorized] +#[post("/{job_id}/cancel")] +async fn cancel_handler( + workspace_context: WorkspaceContext, + state: Data, + db_conn: DbConnection, + job_id: Path, +) -> superposition::Result { + let DbConnection(mut conn) = db_conn; + let job_id_str = job_id.into_inner(); + let job_id: i64 = job_id_str + .parse() + .map_err(|e| unexpected_error!("Invalid job_id '{}': {}", job_id_str, e))?; + + let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); + let job = get_job_by_id(&mut conn, &job_workspace, job_id) + .map_err(|e| unexpected_error!("Failed to fetch job: {}", e))?; + + if matches!( + job.status, + BackgroundJobStatus::Completed | BackgroundJobStatus::Failed + ) { + return Err(unexpected_error!( + "Cannot cancel job {} with terminal status {}", + job_id, + job.status + )); + } + + let target_workspace = state + .kronos_workspace + .clone() + .unwrap_or_else(|| workspace_context.schema_name.to_string()); + + if let Err(e) = state + .kronos_client + .cancel_job(&target_workspace, &job.kronos_job_id) + .await + { + append_job_logs(&mut conn, job_id, &format!("Cancel attempt failed: {e}")) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; + return Err(unexpected_error!("Failed to cancel Kronos job: {}", e)); + } + + update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) + .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; + + append_job_logs(&mut conn, job_id, "Job cancelled by user") + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; + + Ok(HttpResponse::Ok().finish()) +} diff --git a/crates/superposition/src/main.rs b/crates/superposition/src/main.rs index 4c1d2f1f6..e1d41a178 100644 --- a/crates/superposition/src/main.rs +++ b/crates/superposition/src/main.rs @@ -1,6 +1,7 @@ #![deny(unused_crate_dependencies)] mod app_state; mod dispatch; +mod jobs; mod log_span; mod organisation; mod resolve; @@ -468,6 +469,11 @@ impl ScopeExt for Scope { .wrap(OrgWorkspaceMiddlewareFactory::new(true, true)) .service(dispatch::endpoints()), ) + .service( + scope("/jobs") + .wrap(OrgWorkspaceMiddlewareFactory::new(true, true)) + .service(jobs::endpoints()), + ) } fn resource_routes_org_specific(self, auth_z_manager: AuthZManager) -> Self { diff --git a/crates/superposition/src/workspace/handlers.rs b/crates/superposition/src/workspace/handlers.rs index f00df8c31..93b8bee1a 100644 --- a/crates/superposition/src/workspace/handlers.rs +++ b/crates/superposition/src/workspace/handlers.rs @@ -19,7 +19,7 @@ use service_utils::{ rotate_workspace_encryption_key_helper, }, helpers::get_workspace, - kronos_dispatch::setup_dispatcher, + kronos_dispatch::{setup_dispatcher, setup_job_dispatcher}, middlewares::auth_z::AuthZHandler, service::types::{ AppState, DbConnection, OrganisationId, SchemaName, WorkspaceContext, WorkspaceId, @@ -186,6 +186,12 @@ async fn create_handler( &state.kronos_dispatch_token, ) .await; + setup_job_dispatcher( + state.kronos_client.as_ref(), + &workspace_schema_name.0, + &state.superposition_host, + ) + .await; } let _ = authz_handler @@ -480,6 +486,12 @@ async fn migrate_schema_handler( &state.kronos_dispatch_token, ) .await; + setup_job_dispatcher( + state.kronos_client.as_ref(), + &schema_name.0, + &state.superposition_host, + ) + .await; } let workspace = get_workspace(&schema_name, &mut conn)?; diff --git a/crates/superposition_types/src/api/jobs.rs b/crates/superposition_types/src/api/jobs.rs index 0f09f706b..9e12873c7 100644 --- a/crates/superposition_types/src/api/jobs.rs +++ b/crates/superposition_types/src/api/jobs.rs @@ -3,7 +3,8 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::database::models::{ - BackgroundJob, BackgroundJobStatus, BackgroundJobType, + others::WorkspaceJobView, BackgroundJob, BackgroundJobStatus, BackgroundJobType, + JobWorkspace, }; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -48,21 +49,45 @@ impl JobRequest { } #[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobDispatchRequest { + #[serde(with = "crate::database::models::i64_formatter")] + pub job_id: i64, + #[serde(flatten)] + pub job_request: JobRequest, +} + +#[derive(Debug, Clone, Serialize)] pub struct JobResponse { #[serde(with = "crate::database::models::i64_formatter")] pub id: i64, pub kronos_job_id: String, pub description: String, - #[serde(rename = "type")] pub job_type: BackgroundJobType, pub status: BackgroundJobStatus, pub name: String, pub progress: i32, - pub workspace_schema: String, + pub workspace_schema: JobWorkspace, pub created_at: DateTime, pub logs: String, } +impl JobResponse { + pub fn from_view(view: &WorkspaceJobView, schema: &String) -> Self { + Self { + id: view.id, + kronos_job_id: view.kronos_job_id.clone(), + description: view.description.clone(), + job_type: view.job_type, + status: view.status, + name: view.name.clone(), + progress: view.progress, + workspace_schema: JobWorkspace::from(schema), + created_at: view.created_at, + logs: view.logs.clone(), + } + } +} + impl From for JobResponse { fn from(job: BackgroundJob) -> Self { Self { @@ -98,7 +123,7 @@ pub struct ExecutionDetails { pub execution_status: Option, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize)] pub struct JobDetailResponse { #[serde(flatten)] pub job: JobResponse, diff --git a/crates/superposition_types/src/database/models.rs b/crates/superposition_types/src/database/models.rs index a013195b4..4cb11e9e5 100644 --- a/crates/superposition_types/src/database/models.rs +++ b/crates/superposition_types/src/database/models.rs @@ -380,15 +380,20 @@ impl JobWorkspace { Self::Workspace(schema) => schema.clone(), } } +} - pub fn from_workspace_schema(schema: &str) -> Self { - Self::Workspace(schema.to_string()) +impl From<&String> for JobWorkspace { + fn from(value: &String) -> Self { + if value == Self::GLOBAL { + Self::Global + } else { + Self::Workspace(value.clone()) + } } } impl TryFrom for JobWorkspace { type Error = String; - fn try_from(value: String) -> Result { if value == Self::GLOBAL { Ok(Self::Global) diff --git a/crates/superposition_types/src/lib.rs b/crates/superposition_types/src/lib.rs index 6d692200b..a6db05c4f 100644 --- a/crates/superposition_types/src/lib.rs +++ b/crates/superposition_types/src/lib.rs @@ -83,6 +83,8 @@ pub enum Resource { Variable, Secret, MasterEncryptionKey, + WorkspaceJob, + GlobalJob, } impl From for String { From 56a9d45ad1653651b57c1d804c9f08ffc2b91fe2 Mon Sep 17 00:00:00 2001 From: datron Date: Fri, 17 Jul 2026 18:20:56 +0530 Subject: [PATCH 03/12] feat: migrating jobs and API calls Signed-off-by: datron --- crates/context_aware_config/src/api/config.rs | 1 + .../src/api/config/handlers.rs | 62 +++++++---- .../context_aware_config/src/api/context.rs | 1 + .../src/api/context/handlers.rs | 79 ++++++++------ crates/superposition/src/dispatch/handlers.rs | 100 ++++++++++++++++-- 5 files changed, 186 insertions(+), 57 deletions(-) diff --git a/crates/context_aware_config/src/api/config.rs b/crates/context_aware_config/src/api/config.rs index 62a998ddb..18036e2df 100644 --- a/crates/context_aware_config/src/api/config.rs +++ b/crates/context_aware_config/src/api/config.rs @@ -1,3 +1,4 @@ mod handlers; pub use handlers::endpoints; +pub use handlers::execute_reduce; pub mod helpers; diff --git a/crates/context_aware_config/src/api/config/handlers.rs b/crates/context_aware_config/src/api/config/handlers.rs index 3028196f8..28492c162 100644 --- a/crates/context_aware_config/src/api/config/handlers.rs +++ b/crates/context_aware_config/src/api/config/handlers.rs @@ -10,6 +10,7 @@ use itertools::Itertools; use serde_json::{Map, Value, json}; use service_utils::{ helpers::{fetch_dimensions_info_map, is_not_modified}, + kronos_dispatch::submit_job, redis::{CONFIG_KEY_SUFFIX, LAST_MODIFIED_KEY_SUFFIX, read_through_cache}, service::types::{ AppHeader, AppState, DbConnection, WorkspaceContext, WorkspaceWritePermit, @@ -30,6 +31,7 @@ use superposition_types::{ MergeStrategy, ResolveConfigQuery, }, context::PutRequest, + jobs::{JobCreateResponse, JobRequest, ReduceRequest}, }, custom_query::{ self as superposition_query, CustomQuery, DimensionQuery, PaginationParams, @@ -37,7 +39,7 @@ use superposition_types::{ }, database::{ models::{ - ChangeReason, + ChangeReason, JobWorkspace, cac::{ConfigVersion, ConfigVersionListItem}, }, schema::config_versions::dsl as config_versions, @@ -444,22 +446,14 @@ async fn reduce_config_key( }) } -#[authorized] -#[put("/reduce")] -async fn reduce_handler( - workspace_context: WorkspaceContext, - req: HttpRequest, - user: User, +pub async fn execute_reduce( + workspace_context: &WorkspaceContext, mut write_permit: WorkspaceWritePermit, - state: Data, -) -> superposition::Result { + user: &User, + state: &Data, + is_approve: bool, +) -> superposition::Result<()> { let conn = write_permit.connection(); - let is_approve = req - .headers() - .get("x-approve") - .and_then(|value| value.to_str().ok().and_then(|s| s.parse::().ok())) - .unwrap_or(false); - let dimensions_info_map = fetch_dimensions_info_map(conn, &workspace_context.schema_name)?; let mut config = generate_cac(conn, &workspace_context.schema_name)?; @@ -477,16 +471,48 @@ async fn reduce_handler( &dimensions_info_map, default_config.clone(), is_approve, - &workspace_context, - &state, + workspace_context, + state, ) .await?; if is_approve { config = generate_cac(conn, &workspace_context.schema_name)?; } } + Ok(()) +} + +#[authorized] +#[put("/reduce")] +async fn reduce_handler( + workspace_context: WorkspaceContext, + db_conn: DbConnection, + state: Data, +) -> superposition::Result> { + let DbConnection(mut conn) = db_conn; + let job_request = JobRequest::Reduce(ReduceRequest::default()); + let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); + let target_workspace = state + .kronos_workspace + .as_deref() + .unwrap_or(&workspace_context.schema_name); + + let response = submit_job( + state.kronos_client.as_ref(), + target_workspace, + &job_workspace, + &workspace_context.organisation_id, + &workspace_context.workspace_id, + &job_request, + &state.snowflake_generator, + &mut conn, + 3, + "Reduce config job", + ) + .await + .map_err(|e| unexpected_error!("Failed to submit reduce job: {}", e))?; - Ok(HttpResponse::Ok().json(config)) + Ok(Json(response)) } #[authorized] diff --git a/crates/context_aware_config/src/api/context.rs b/crates/context_aware_config/src/api/context.rs index 0d6cee04d..d94597aa2 100644 --- a/crates/context_aware_config/src/api/context.rs +++ b/crates/context_aware_config/src/api/context.rs @@ -4,6 +4,7 @@ pub mod operations; mod types; pub mod validations; pub use handlers::endpoints; +pub use handlers::execute_priority_recompute; pub use operations::delete; pub use operations::update; pub use operations::upsert; diff --git a/crates/context_aware_config/src/api/context/handlers.rs b/crates/context_aware_config/src/api/context/handlers.rs index 09f4a0cc6..f66379e25 100644 --- a/crates/context_aware_config/src/api/context/handlers.rs +++ b/crates/context_aware_config/src/api/context/handlers.rs @@ -17,6 +17,7 @@ use service_utils::{ helpers::{ WebhookData, execute_webhook_call, fetch_dimensions_info_map, parse_config_tags, }, + kronos_dispatch::submit_job, middlewares::auth_z::{Action as AuthZAction, AuthZ}, service::types::{ AppHeader, AppState, CustomHeaders, DbConnection, SchemaName, WorkspaceContext, @@ -27,8 +28,8 @@ use superposition_core::helpers::{calculate_context_weight, hash}; use superposition_derives::{authorized, declare_resource}; use superposition_macros::{bad_argument, db_error, unexpected_error}; use superposition_types::{ - Contextual, DBConnection, DimensionInfo, InternalUserContext, ListResponse, - Overridden, Overrides, PaginatedResponse, PrefixList, Resource, SortBy, User, + Contextual, DBConnection, DimensionInfo, InternalUserContext, Overridden, Overrides, + PaginatedResponse, Resource, SortBy, User, api::{ DimensionMatchStrategy, context::{ @@ -36,6 +37,7 @@ use superposition_types::{ ContextListFilters, ContextValidationRequest, Identifier, MoveRequest, PutRequest, SortOn, UpdateRequest, WeightRecomputeResponse, }, + jobs::{JobCreateResponse, JobRequest, PriorityRecomputeRequest}, webhook::Action, }, custom_query::{ @@ -43,7 +45,9 @@ use superposition_types::{ QueryMap, }, database::{ - models::{ChangeReason, Description, cac::Context, others::WebhookEvent}, + models::{ + ChangeReason, Description, JobWorkspace, cac::Context, others::WebhookEvent, + }, schema::contexts::{self, dsl, id}, }, logic::evaluate_local_cohorts_skip_unresolved, @@ -1161,15 +1165,12 @@ async fn bulk_operations_handler( Ok(http_resp) } -#[authorized] -#[put("/weight/recompute")] -async fn weight_recompute_handler( - workspace_context: WorkspaceContext, - state: Data, - custom_headers: CustomHeaders, +pub async fn execute_priority_recompute( + workspace_context: &WorkspaceContext, + state: &Data, mut write_permit: WorkspaceWritePermit, - user: User, -) -> superposition::Result { + user: &User, +) -> superposition::Result<()> { use superposition_types::database::schema::contexts::dsl::{ contexts, last_modified_at, last_modified_by, weight, }; @@ -1187,7 +1188,6 @@ async fn weight_recompute_handler( let dimension_info_map = fetch_dimensions_info_map(conn, &workspace_context.schema_name)?; let mut response: Vec = vec![]; - let tags = parse_config_tags(custom_headers.config_tags)?; let contexts_new_weight = result .clone() @@ -1214,7 +1214,6 @@ async fn weight_recompute_handler( }) .collect::>>()?; - // Update database and add config version let last_modified_time = Utc::now(); let config_version = conn.transaction::<_, superposition::AppError, _>(|transaction_conn| { @@ -1234,8 +1233,9 @@ async fn weight_recompute_handler( db_error!(err) })?; } - let config_version_desc = Description::try_from("Recomputed weight".to_string()).map_err(|e| unexpected_error!(e))?; - add_config_version(&state, tags, config_version_desc, transaction_conn, &workspace_context.schema_name) + let config_version_desc = Description::try_from("Recomputed weight".to_string()) + .map_err(|e| unexpected_error!(e))?; + add_config_version(state, None, config_version_desc, transaction_conn, &workspace_context.schema_name) })?; let _ = put_config_in_redis( &config_version, @@ -1253,22 +1253,41 @@ async fn weight_recompute_handler( action: Action::Batch(vec![Action::Update; response.len()]), }; - let webhook_status = - execute_webhook_call(data, &workspace_context, &state, conn).await; + let _ = execute_webhook_call(data, workspace_context, state, conn).await; + Ok(()) +} - let mut http_resp = if webhook_status { - HttpResponse::Ok() - } else { - HttpResponse::build( - actix_web::http::StatusCode::from_u16(512) - .unwrap_or(actix_web::http::StatusCode::INTERNAL_SERVER_ERROR), - ) - }; - http_resp.insert_header(( - AppHeader::XConfigVersion.to_string(), - config_version.id.to_string(), - )); - Ok(http_resp.json(ListResponse::new(response))) +#[authorized] +#[put("/weight/recompute")] +async fn weight_recompute_handler( + workspace_context: WorkspaceContext, + state: Data, + db_conn: DbConnection, +) -> superposition::Result> { + let DbConnection(mut conn) = db_conn; + let job_request = JobRequest::PriorityRecompute(PriorityRecomputeRequest::default()); + let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); + let target_workspace = state + .kronos_workspace + .as_deref() + .unwrap_or(&workspace_context.schema_name); + + let response = submit_job( + state.kronos_client.as_ref(), + target_workspace, + &job_workspace, + &workspace_context.organisation_id, + &workspace_context.workspace_id, + &job_request, + &state.snowflake_generator, + &mut conn, + 3, + "Priority recompute job", + ) + .await + .map_err(|e| unexpected_error!("Failed to submit priority recompute job: {}", e))?; + + Ok(Json(response)) } #[authorized] diff --git a/crates/superposition/src/dispatch/handlers.rs b/crates/superposition/src/dispatch/handlers.rs index d7124f48c..7fad43d48 100644 --- a/crates/superposition/src/dispatch/handlers.rs +++ b/crates/superposition/src/dispatch/handlers.rs @@ -4,18 +4,27 @@ use actix_web::{ HttpResponse, Scope, post, web::{Data, Json}, }; +use context_aware_config::api::{ + config::execute_reduce, context::execute_priority_recompute, +}; use diesel::{QueryDsl, RunQueryDsl}; use secrecy::ExposeSecret; use serde::Deserialize; use service_utils::{ encryption::{EncryptionError, decrypt_secret, decrypt_workspace_key}, helpers::get_from_env_or_default, - kronos_dispatch::{has_pattern_in_headers, substitute_templates}, + kronos_dispatch::{ + append_job_logs, has_pattern_in_headers, substitute_templates, + update_job_progress, update_job_status, + }, service::types::{AppState, DbConnection, WorkspaceContext}, }; use superposition_derives::{authorized, declare_resource}; use superposition_macros::unexpected_error; use superposition_types::{ + User, + api::jobs::{JobDispatchRequest, JobRequest}, + database::models::BackgroundJobStatus, database::schema::{secrets::dsl as secrets_dsl, variables::dsl as variables_dsl}, result as superposition, }; @@ -31,7 +40,9 @@ struct DispatchWebhookRequest { } pub fn endpoints() -> Scope { - Scope::new("").service(dispatch_handler) + Scope::new("") + .service(dispatch_handler) + .service(dispatch_job_handler) } #[authorized] @@ -43,26 +54,97 @@ async fn dispatch_handler( body: Json, ) -> superposition::Result { let DispatchWebhookRequest { webhook_name, data } = body.into_inner(); + let DbConnection(mut conn) = db_conn; + + execute_webhook_dispatch(&workspace_context, &state, &mut conn, &webhook_name, &data) + .await?; + Ok(HttpResponse::Ok().finish()) +} +#[authorized] +#[post("/job")] +async fn dispatch_job_handler( + workspace_context: WorkspaceContext, + state: Data, + db_conn: DbConnection, + user: User, + body: Json, +) -> superposition::Result { + let JobDispatchRequest { + job_id, + job_request, + } = body.into_inner(); let DbConnection(mut conn) = db_conn; - let webhook = - fetch_webhook(&webhook_name, &workspace_context.schema_name, &mut conn)?; + + update_job_status(&mut conn, job_id, BackgroundJobStatus::Inprogress) + .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; + + let result = match &job_request { + JobRequest::Webhook(req) => { + execute_webhook_dispatch( + &workspace_context, + &state, + &mut conn, + &req.webhook_name, + &req.data, + ) + .await + } + JobRequest::PriorityRecompute(_) => { + execute_priority_recompute(&workspace_context, &state, &mut conn, &user).await + } + JobRequest::Reduce(req) => { + execute_reduce(&workspace_context, &state, &mut conn, &user, req.approve) + .await + } + }; + + match result { + Ok(()) => { + update_job_status(&mut conn, job_id, BackgroundJobStatus::Completed) + .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; + update_job_progress(&mut conn, job_id, 100) + .map_err(|e| unexpected_error!("Failed to update job progress: {}", e))?; + Ok(HttpResponse::Ok().finish()) + } + Err(e) => { + let error_msg = format!("{e}"); + log::error!("Job {job_id} failed: {error_msg}"); + update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) + .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; + append_job_logs(&mut conn, job_id, &error_msg) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; + Err(unexpected_error!("Job {job_id} failed: {error_msg}")) + } + } +} + +async fn execute_webhook_dispatch( + workspace_context: &WorkspaceContext, + state: &Data, + conn: &mut diesel::r2d2::PooledConnection< + diesel::r2d2::ConnectionManager, + >, + webhook_name: &String, + data: &serde_json::Value, +) -> superposition::Result<()> { + let webhook = fetch_webhook(webhook_name, &workspace_context.schema_name, conn)?; if !webhook.enabled { - return Ok(HttpResponse::Ok().finish()); + return Ok(()); } let raw_headers = (*webhook.custom_headers).clone(); let (has_vars, has_secrets) = has_pattern_in_headers(&raw_headers); let vars = if has_vars { - fetch_variables(&workspace_context, &mut conn)? + fetch_variables(workspace_context, conn)? } else { HashMap::new() }; let secrets = if has_secrets { - fetch_decrypted_secrets(&workspace_context, &mut conn, &state)? + fetch_decrypted_secrets(workspace_context, conn, state)? } else { HashMap::new() }; @@ -78,7 +160,7 @@ async fn dispatch_handler( "WEBHOOK_OUTBOUND_TIMEOUT_SEC", 10u64, ))) - .json(&data); + .json(data); for (key, value) in &raw_headers { let value_str = value @@ -95,7 +177,7 @@ async fn dispatch_handler( .map_err(|e| unexpected_error!("Dispatcher HTTP send failed: {}", e))?; if resp.status().is_success() { - Ok(HttpResponse::Ok().finish()) + Ok(()) } else { Err(unexpected_error!( "Target returned unexpected status {}", From a46b48d97a2e89f45b23c564efada354885b50cf Mon Sep 17 00:00:00 2001 From: datron Date: Mon, 20 Jul 2026 13:22:13 +0530 Subject: [PATCH 04/12] fix: add missing DB queries and include kronos setup sql Signed-off-by: datron --- docker-compose/postgres/db_init.sql | 81 +++++++++++++++ kronos_setup.sql | 155 ++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+) create mode 100644 kronos_setup.sql diff --git a/docker-compose/postgres/db_init.sql b/docker-compose/postgres/db_init.sql index b28268fb6..9ebc3fe5a 100644 --- a/docker-compose/postgres/db_init.sql +++ b/docker-compose/postgres/db_init.sql @@ -1783,4 +1783,85 @@ ALTER TABLE localorg_test.experiments ADD COLUMN IF NOT EXISTS idempotency_key T CREATE UNIQUE INDEX IF NOT EXISTS experiments_idempotency_key_idx ON localorg_dev.experiments(idempotency_key) WHERE idempotency_key IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS experiments_idempotency_key_idx ON localorg_test.experiments(idempotency_key) WHERE idempotency_key IS NOT NULL; +DO $$ BEGIN + CREATE TYPE public.background_job_type AS ENUM ( + 'WEBHOOK', + 'PRIORITY_RECOMPUTE', + 'REDUCE' + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +DO $$ BEGIN + CREATE TYPE public.background_job_status AS ENUM ( + 'CREATED', + 'SCHEDULED', + 'INPROGRESS', + 'FAILED', + 'COMPLETED' + ); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; + +CREATE TABLE IF NOT EXISTS superposition.job_manager ( + id BIGINT PRIMARY KEY, + kronos_job_id TEXT NOT NULL, + description TEXT NOT NULL, + job_type public.background_job_type NOT NULL, + status public.background_job_status NOT NULL, + name TEXT NOT NULL, + progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100), + workspace_schema TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + logs TEXT NOT NULL DEFAULT '' +); + +CREATE INDEX IF NOT EXISTS idx_job_manager_workspace_schema + ON superposition.job_manager (workspace_schema); + +CREATE INDEX IF NOT EXISTS idx_job_manager_kronos_job_id + ON superposition.job_manager (kronos_job_id); + +CREATE INDEX IF NOT EXISTS idx_job_manager_type + ON superposition.job_manager (job_type); + +CREATE INDEX IF NOT EXISTS idx_job_manager_status + ON superposition.job_manager (status); + +CREATE INDEX IF NOT EXISTS idx_job_manager_status_job_type + ON superposition.job_manager (status, job_type); + +CREATE INDEX IF NOT EXISTS idx_job_manager_created_at + ON superposition.job_manager (created_at DESC); + +CREATE OR REPLACE VIEW localorg_dev.job_manager AS +SELECT + id, + kronos_job_id, + description, + job_type, + status, + name, + progress, + created_at, + logs +FROM superposition.job_manager +WHERE workspace_schema = 'localorg_dev'; + +CREATE OR REPLACE VIEW localorg_test.job_manager AS +SELECT + id, + kronos_job_id, + description, + job_type, + status, + name, + progress, + created_at, + logs +FROM superposition.job_manager +WHERE workspace_schema = 'localorg_test'; + COMMIT; diff --git a/kronos_setup.sql b/kronos_setup.sql new file mode 100644 index 000000000..049cd9b29 --- /dev/null +++ b/kronos_setup.sql @@ -0,0 +1,155 @@ +-- use this only when kronos is in library mode + +CREATE EXTENSION IF NOT EXISTS pgcrypto; + +CREATE TABLE IF NOT EXISTS kronos_payload_specs ( + name TEXT NOT NULL, + schema_json JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_payload_specs PRIMARY KEY (name) +); + +CREATE TABLE IF NOT EXISTS kronos_configs ( + name TEXT NOT NULL, + values_json JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_configs PRIMARY KEY (name) +); + +CREATE TABLE IF NOT EXISTS kronos_secrets ( + name TEXT NOT NULL, + encrypted_value BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_secrets PRIMARY KEY (name) +); + +CREATE TABLE IF NOT EXISTS kronos_endpoints ( + name TEXT NOT NULL, + endpoint_type TEXT NOT NULL, + payload_spec_ref TEXT, + config_ref TEXT, + spec JSONB NOT NULL, + retry_policy JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_endpoints PRIMARY KEY (name), + CONSTRAINT fk_kronos_endpoints_payload_spec FOREIGN KEY (payload_spec_ref) REFERENCES kronos_payload_specs (name), + CONSTRAINT fk_kronos_endpoints_config FOREIGN KEY (config_ref) REFERENCES kronos_configs (name), + CONSTRAINT chk_kronos_endpoint_type CHECK (endpoint_type IN ('HTTP', 'KAFKA', 'REDIS_STREAM', 'INTERNAL')) +); + +CREATE INDEX IF NOT EXISTS idx_kronos_endpoints_type ON kronos_endpoints (endpoint_type); + +CREATE TABLE IF NOT EXISTS kronos_jobs ( + job_id TEXT NOT NULL DEFAULT gen_random_uuid()::TEXT, + endpoint TEXT NOT NULL, + endpoint_type TEXT NOT NULL, + trigger_type TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'ACTIVE', + version BIGINT NOT NULL DEFAULT 1, + previous_version_id TEXT, + replaced_by_id TEXT, + idempotency_key TEXT, + input JSONB, + run_at TIMESTAMPTZ, + cron_expression TEXT, + cron_timezone TEXT, + cron_starts_at TIMESTAMPTZ, + cron_ends_at TIMESTAMPTZ, + cron_next_run_at TIMESTAMPTZ, + cron_last_tick_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + retired_at TIMESTAMPTZ, + CONSTRAINT pk_kronos_jobs PRIMARY KEY (job_id), + CONSTRAINT fk_kronos_jobs_endpoint FOREIGN KEY (endpoint) REFERENCES kronos_endpoints (name), + CONSTRAINT chk_kronos_trigger_type CHECK (trigger_type IN ('IMMEDIATE', 'DELAYED', 'CRON')), + CONSTRAINT chk_kronos_job_status CHECK (status IN ('ACTIVE', 'RETIRED')), + CONSTRAINT chk_kronos_job_endpoint_type CHECK (endpoint_type IN ('HTTP', 'KAFKA', 'REDIS_STREAM', 'INTERNAL')) +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_kronos_jobs_idempotency + ON kronos_jobs (endpoint, idempotency_key) + WHERE idempotency_key IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_kronos_jobs_cron_due + ON kronos_jobs (cron_next_run_at) + WHERE trigger_type = 'CRON' AND status = 'ACTIVE'; + +CREATE INDEX IF NOT EXISTS idx_kronos_jobs_endpoint ON kronos_jobs (endpoint, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_kronos_jobs_status ON kronos_jobs (status, created_at DESC); + +CREATE TABLE IF NOT EXISTS kronos_executions ( + execution_id TEXT NOT NULL DEFAULT gen_random_uuid()::TEXT, + job_id TEXT NOT NULL, + endpoint TEXT NOT NULL, + endpoint_type TEXT NOT NULL, + idempotency_key TEXT, + status TEXT NOT NULL DEFAULT 'PENDING', + input JSONB, + output JSONB, + attempt_count BIGINT NOT NULL DEFAULT 0, + max_attempts BIGINT NOT NULL DEFAULT 1, + worker_id TEXT, + run_at TIMESTAMPTZ NOT NULL DEFAULT now(), + started_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + duration_ms BIGINT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_executions PRIMARY KEY (execution_id), + CONSTRAINT fk_kronos_executions_job FOREIGN KEY (job_id) REFERENCES kronos_jobs (job_id), + CONSTRAINT chk_kronos_exec_status CHECK (status IN ( + 'PENDING', 'QUEUED', 'RUNNING', 'RETRYING', 'SUCCESS', 'FAILED', 'CANCELLED' + )) +); + +CREATE INDEX IF NOT EXISTS idx_kronos_executions_pickup + ON kronos_executions (status, run_at ASC) + WHERE status IN ('QUEUED', 'RETRYING', 'PENDING'); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_kronos_executions_cron_dedup + ON kronos_executions (job_id, idempotency_key) + WHERE idempotency_key IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_kronos_executions_by_job ON kronos_executions (job_id, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_kronos_executions_running ON kronos_executions (status, started_at) + WHERE status = 'RUNNING'; + +CREATE TABLE IF NOT EXISTS kronos_attempts ( + attempt_id TEXT NOT NULL DEFAULT gen_random_uuid()::TEXT, + execution_id TEXT NOT NULL, + attempt_number BIGINT NOT NULL, + status TEXT NOT NULL, + started_at TIMESTAMPTZ NOT NULL, + completed_at TIMESTAMPTZ, + duration_ms BIGINT, + output JSONB, + error JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_attempts PRIMARY KEY (attempt_id), + CONSTRAINT fk_kronos_attempts_execution FOREIGN KEY (execution_id) REFERENCES kronos_executions (execution_id), + CONSTRAINT uq_kronos_attempts_exec_number UNIQUE (execution_id, attempt_number), + CONSTRAINT chk_kronos_attempt_status CHECK (status IN ('SUCCESS', 'FAILED')) +); + +CREATE INDEX IF NOT EXISTS idx_kronos_attempts_by_execution + ON kronos_attempts (execution_id, attempt_number ASC); + +CREATE TABLE IF NOT EXISTS kronos_execution_logs ( + log_id TEXT NOT NULL DEFAULT gen_random_uuid()::TEXT, + execution_id TEXT NOT NULL, + attempt_number BIGINT NOT NULL, + level TEXT NOT NULL, + message TEXT NOT NULL, + logged_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT pk_kronos_execution_logs PRIMARY KEY (log_id), + CONSTRAINT fk_kronos_logs_execution FOREIGN KEY (execution_id) REFERENCES kronos_executions (execution_id), + CONSTRAINT chk_kronos_log_level CHECK (level IN ('DEBUG', 'INFO', 'WARN', 'ERROR')) +); + +CREATE INDEX IF NOT EXISTS idx_kronos_logs_by_execution + ON kronos_execution_logs (execution_id, logged_at ASC); +CREATE INDEX IF NOT EXISTS idx_kronos_logs_by_attempt + ON kronos_execution_logs (execution_id, attempt_number, logged_at ASC); From 88ba3e30a46e092d16740d5e03565cafad87e699 Mon Sep 17 00:00:00 2001 From: datron Date: Mon, 20 Jul 2026 18:16:48 +0530 Subject: [PATCH 05/12] fix: make logs jsonb keyed against a time stamp Signed-off-by: datron --- crates/service_utils/src/kronos_dispatch.rs | 42 ++++++++++++++----- crates/superposition/src/dispatch/handlers.rs | 6 ++- crates/superposition/src/jobs/handlers.rs | 13 ++++-- .../2026-07-16-000001_job_manager/up.sql | 2 +- crates/superposition_types/src/api/jobs.rs | 4 +- .../src/database/models.rs | 2 +- .../src/database/models/others.rs | 2 +- .../src/database/schema.rs | 2 +- .../src/database/superposition_schema.rs | 2 +- docker-compose/postgres/db_init.sql | 2 +- superposition.sql | 2 +- 11 files changed, 55 insertions(+), 24 deletions(-) diff --git a/crates/service_utils/src/kronos_dispatch.rs b/crates/service_utils/src/kronos_dispatch.rs index cf3af7af4..b917911af 100644 --- a/crates/service_utils/src/kronos_dispatch.rs +++ b/crates/service_utils/src/kronos_dispatch.rs @@ -289,7 +289,7 @@ pub async fn submit_job( progress: 0, workspace_schema: workspace.clone(), created_at: Utc::now(), - logs: String::new(), + logs: json!({}), }; diesel::insert_into(job_manager_dsl::job_manager) @@ -342,12 +342,18 @@ pub async fn submit_job( Err(e) => { let error_msg = format!("Kronos job creation failed: {e}"); log::error!("submit_job: {error_msg}"); + let mut log_map = serde_json::Map::new(); + log_map.insert( + Utc::now().to_rfc3339(), + serde_json::Value::String(error_msg), + ); + let log_entry = serde_json::Value::Object(log_map); diesel::update( job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id)), ) .set(( job_manager_dsl::status.eq(BackgroundJobStatus::Failed), - job_manager_dsl::logs.eq(&error_msg), + job_manager_dsl::logs.eq(&log_entry), )) .execute(conn)?; Err(e) @@ -410,9 +416,19 @@ pub fn update_job_progress( job_id: i64, progress: i32, ) -> anyhow::Result<()> { - diesel::update(job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id))) + let threshold: i32 = get_from_env_or_default("JOB_PROGRESS_DIFF_UPDATE", 10); + let previous_progress = job_manager_dsl::job_manager + .filter(job_manager_dsl::id.eq(job_id)) + .select(job_manager_dsl::progress) + .first::(conn)?; + + if (progress - previous_progress).abs() > threshold { + diesel::update( + job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id)), + ) .set(job_manager_dsl::progress.eq(progress)) .execute(conn)?; + } Ok(()) } @@ -420,20 +436,24 @@ pub fn append_job_logs( conn: &mut DBConnection, job_id: i64, log_line: &str, -) -> anyhow::Result<()> { + key: Option, +) -> anyhow::Result { let current = job_manager_dsl::job_manager .filter(job_manager_dsl::id.eq(job_id)) .select(job_manager_dsl::logs) - .first::(conn)?; + .first::(conn)?; - let new_logs = if current.is_empty() { - log_line.to_string() - } else { - format!("{current}\n{log_line}") - }; + let timed_key = key.unwrap_or(Utc::now().to_rfc3339()); + + let mut logs = current.as_object().cloned().unwrap_or_default(); + logs.insert( + timed_key.clone(), + serde_json::Value::String(log_line.to_string()), + ); + let new_logs = serde_json::Value::Object(logs); diesel::update(job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id))) .set(job_manager_dsl::logs.eq(new_logs)) .execute(conn)?; - Ok(()) + Ok(timed_key) } diff --git a/crates/superposition/src/dispatch/handlers.rs b/crates/superposition/src/dispatch/handlers.rs index 7fad43d48..52ffd1adb 100644 --- a/crates/superposition/src/dispatch/handlers.rs +++ b/crates/superposition/src/dispatch/handlers.rs @@ -78,6 +78,8 @@ async fn dispatch_job_handler( update_job_status(&mut conn, job_id, BackgroundJobStatus::Inprogress) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; + let key = append_job_logs(&mut conn, job_id, "Job started", None) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; let result = match &job_request { JobRequest::Webhook(req) => { @@ -105,6 +107,8 @@ async fn dispatch_job_handler( .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; update_job_progress(&mut conn, job_id, 100) .map_err(|e| unexpected_error!("Failed to update job progress: {}", e))?; + append_job_logs(&mut conn, job_id, "Job completed", Some(key)) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; Ok(HttpResponse::Ok().finish()) } Err(e) => { @@ -112,7 +116,7 @@ async fn dispatch_job_handler( log::error!("Job {job_id} failed: {error_msg}"); update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; - append_job_logs(&mut conn, job_id, &error_msg) + append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), Some(key)) .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; Err(unexpected_error!("Job {job_id} failed: {error_msg}")) } diff --git a/crates/superposition/src/jobs/handlers.rs b/crates/superposition/src/jobs/handlers.rs index c4c7e29e9..1a6f6405d 100644 --- a/crates/superposition/src/jobs/handlers.rs +++ b/crates/superposition/src/jobs/handlers.rs @@ -123,20 +123,27 @@ async fn cancel_handler( .clone() .unwrap_or_else(|| workspace_context.schema_name.to_string()); + let key = chrono::Utc::now().to_rfc3339(); + if let Err(e) = state .kronos_client .cancel_job(&target_workspace, &job.kronos_job_id) .await { - append_job_logs(&mut conn, job_id, &format!("Cancel attempt failed: {e}")) - .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; + let _ = append_job_logs( + &mut conn, + job_id, + &format!("Cancel attempt failed: {e}"), + Some(key.clone()), + ) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; return Err(unexpected_error!("Failed to cancel Kronos job: {}", e)); } update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; - append_job_logs(&mut conn, job_id, "Job cancelled by user") + let _ = append_job_logs(&mut conn, job_id, "Job cancelled by user", Some(key)) .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; Ok(HttpResponse::Ok().finish()) diff --git a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql index f641f27bb..c905367e6 100644 --- a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql +++ b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql @@ -30,7 +30,7 @@ CREATE TABLE IF NOT EXISTS superposition.job_manager ( progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100), workspace_schema TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - logs TEXT NOT NULL DEFAULT '' + logs JSONB NOT NULL DEFAULT '{}'::jsonb ); CREATE INDEX IF NOT EXISTS idx_job_manager_workspace_schema diff --git a/crates/superposition_types/src/api/jobs.rs b/crates/superposition_types/src/api/jobs.rs index 9e12873c7..b08208913 100644 --- a/crates/superposition_types/src/api/jobs.rs +++ b/crates/superposition_types/src/api/jobs.rs @@ -22,7 +22,7 @@ pub struct DispatchWebhookRequest { } #[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct PriorityRecomputeRequest {} +pub struct PriorityRecomputeRequest; #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct ReduceRequest { @@ -68,7 +68,7 @@ pub struct JobResponse { pub progress: i32, pub workspace_schema: JobWorkspace, pub created_at: DateTime, - pub logs: String, + pub logs: serde_json::Value, } impl JobResponse { diff --git a/crates/superposition_types/src/database/models.rs b/crates/superposition_types/src/database/models.rs index 4cb11e9e5..6cce096f0 100644 --- a/crates/superposition_types/src/database/models.rs +++ b/crates/superposition_types/src/database/models.rs @@ -438,7 +438,7 @@ pub struct BackgroundJob { pub progress: i32, pub workspace_schema: JobWorkspace, pub created_at: DateTime, - pub logs: String, + pub logs: serde_json::Value, } #[derive(Clone, Serialize, Deserialize, Debug)] diff --git a/crates/superposition_types/src/database/models/others.rs b/crates/superposition_types/src/database/models/others.rs index f0486c1a3..d1257b955 100644 --- a/crates/superposition_types/src/database/models/others.rs +++ b/crates/superposition_types/src/database/models/others.rs @@ -287,5 +287,5 @@ pub struct WorkspaceJobView { pub name: String, pub progress: i32, pub created_at: DateTime, - pub logs: String, + pub logs: serde_json::Value, } diff --git a/crates/superposition_types/src/database/schema.rs b/crates/superposition_types/src/database/schema.rs index e3036bd69..a4b662c17 100644 --- a/crates/superposition_types/src/database/schema.rs +++ b/crates/superposition_types/src/database/schema.rs @@ -255,7 +255,7 @@ diesel::table! { name -> Text, progress -> Int4, created_at -> Timestamptz, - logs -> Text, + logs -> Jsonb, } } diff --git a/crates/superposition_types/src/database/superposition_schema.rs b/crates/superposition_types/src/database/superposition_schema.rs index 721fd5b0f..4136db861 100644 --- a/crates/superposition_types/src/database/superposition_schema.rs +++ b/crates/superposition_types/src/database/superposition_schema.rs @@ -93,7 +93,7 @@ pub mod superposition { progress -> Int4, workspace_schema -> Text, created_at -> Timestamptz, - logs -> Text, + logs -> Jsonb, } } diff --git a/docker-compose/postgres/db_init.sql b/docker-compose/postgres/db_init.sql index 9ebc3fe5a..2338c4204 100644 --- a/docker-compose/postgres/db_init.sql +++ b/docker-compose/postgres/db_init.sql @@ -1815,7 +1815,7 @@ CREATE TABLE IF NOT EXISTS superposition.job_manager ( progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100), workspace_schema TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - logs TEXT NOT NULL DEFAULT '' + logs JSONB NOT NULL DEFAULT '{}'::jsonb ); CREATE INDEX IF NOT EXISTS idx_job_manager_workspace_schema diff --git a/superposition.sql b/superposition.sql index 0a0685ad4..dad490df9 100644 --- a/superposition.sql +++ b/superposition.sql @@ -187,7 +187,7 @@ CREATE TABLE IF NOT EXISTS superposition.job_manager ( progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100), workspace_schema TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - logs TEXT NOT NULL DEFAULT '' + logs JSONB NOT NULL DEFAULT '{}'::jsonb ); CREATE INDEX IF NOT EXISTS idx_job_manager_workspace_schema From 6a8ffc05ad6a06bdf244fc23218d98f32b213b0f Mon Sep 17 00:00:00 2001 From: datron Date: Tue, 21 Jul 2026 14:03:36 +0530 Subject: [PATCH 06/12] feat: add smithy definitions Signed-off-by: datron --- smithy/models/config.smithy | 10 ++ smithy/models/context.smithy | 7 +- smithy/models/jobs.smithy | 185 +++++++++++++++++++++++++++++++++++ smithy/models/main.smithy | 1 + smithy/smithy-build.json | 3 +- 5 files changed, 200 insertions(+), 6 deletions(-) create mode 100644 smithy/models/jobs.smithy diff --git a/smithy/models/config.smithy b/smithy/models/config.smithy index 59623fcb5..872124c3f 100644 --- a/smithy/models/config.smithy +++ b/smithy/models/config.smithy @@ -23,6 +23,7 @@ resource Config { GetResolvedConfigWithIdentifier GetConfigToml GetConfigJson + Reduce ] } @@ -560,3 +561,12 @@ operation GetResolvedConfigWithIdentifier { audit_id: String } } + +@documentation("Reduces the configuration by removing redundant overrides across contexts. This operation is asynchronous — it submits a background job and returns the job ID for polling.") +@http(method: "PUT", uri: "/config/reduce") +@tags(["Configuration Management"]) +operation Reduce { + input := with [WorkspaceMixin] {} + + output: JobCreateResponse +} diff --git a/smithy/models/context.smithy b/smithy/models/context.smithy index f392b171e..afdc346b4 100644 --- a/smithy/models/context.smithy +++ b/smithy/models/context.smithy @@ -281,7 +281,7 @@ list WeightRecomputeResponses { member: WeightRecomputeResponse } -@documentation("Recalculates and updates the priority weights for all contexts in the workspace based on their dimensions.") +@documentation("Recalculates and updates the priority weights for all contexts in the workspace based on their dimensions. This operation is asynchronous — it submits a background job and returns the job ID for polling.") @http(method: "PUT", uri: "/context/weight/recompute") @tags(["Context Management"]) operation WeightRecompute with [WebhookOperation, WorkspaceWriteOperation] { @@ -291,10 +291,7 @@ operation WeightRecompute with [WebhookOperation, WorkspaceWriteOperation] { config_tags: String } - output := { - @notProperty - data: WeightRecomputeResponses - } + output: JobCreateResponse } structure ContextPut for Context { diff --git a/smithy/models/jobs.smithy b/smithy/models/jobs.smithy new file mode 100644 index 000000000..6796503ff --- /dev/null +++ b/smithy/models/jobs.smithy @@ -0,0 +1,185 @@ +$version: "2.0" + +namespace io.superposition + +resource Job { + identifiers: { + workspace_id: String + org_id: String + id: String + } + properties: { + kronos_job_id: String + description: String + type: BackgroundJobType + status: BackgroundJobStatus + name: String + progress: Integer + workspace_schema: String + created_at: DateTime + logs: Document + } + read: GetJob + list: ListJobs + operations: [ + CancelJob + ] +} + +@documentation("Type of background job.") +enum BackgroundJobType { + WEBHOOK + PRIORITY_RECOMPUTE + REDUCE +} + +@documentation("Lifecycle status of a background job.") +enum BackgroundJobStatus { + CREATED + SCHEDULED + INPROGRESS + FAILED + COMPLETED +} + +@documentation("Execution details fetched from Kronos for a job.") +structure ExecutionDetails { + attempt_count: Long + + max_attempts: Long + + started_at: DateTime + + completed_at: DateTime + + duration_ms: Long + + execution_status: String +} + +@documentation("Full job detail including Kronos execution information.") +structure JobDetailResponse for Job { + @required + $id + + @required + $kronos_job_id + + @required + $description + + @required + $type + + @required + $status + + @required + $name + + @required + $progress + + @required + $workspace_schema + + @required + $created_at + + @required + $logs + + execution: ExecutionDetails +} + +@documentation("Job summary returned by list operations. Does not include workspace_schema.") +structure JobSummary for Job { + @required + $id + + @required + $kronos_job_id + + @required + $description + + @required + $type + + @required + $status + + @required + $name + + @required + $progress + + @required + $created_at + + @required + $logs +} + +list JobList { + member: JobSummary +} + +@documentation("Response returned when a job is submitted. Contains the BJM job ID, Kronos job ID, and initial status.") +structure JobCreateResponse for Job { + @required + $id + + @required + $kronos_job_id + + @required + $status +} + +@documentation("Retrieves a paginated list of background jobs in the workspace, optionally filtered by type and status.") +@readonly +@http(method: "GET", uri: "/jobs") +@tags(["Background Jobs"]) +operation ListJobs { + input := with [WorkspaceMixin] { + @httpQuery("status") + @notProperty + status: BackgroundJobStatus + + @httpQuery("job_type") + @notProperty + job_type: BackgroundJobType + } + + output := { + @required + data: JobList + } +} + +@documentation("Retrieves detailed information about a specific background job, including Kronos execution details such as attempt count, timing, and duration.") +@readonly +@http(method: "GET", uri: "/jobs/{id}") +@tags(["Background Jobs"]) +operation GetJob with [GetOperation] { + input := for Job with [WorkspaceMixin] { + @httpLabel + @required + $id + } + + output: JobDetailResponse +} + +@documentation("Cancels a background job that is not in a terminal state (COMPLETED or FAILED). Sends a cancellation request to Kronos and marks the job as FAILED.") +@http(method: "POST", uri: "/jobs/{id}/cancel") +@tags(["Background Jobs"]) +operation CancelJob with [GetOperation] { + input := for Job with [WorkspaceMixin] { + @httpLabel + @required + $id + } +} diff --git a/smithy/models/main.smithy b/smithy/models/main.smithy index e67e344d5..f61f79562 100644 --- a/smithy/models/main.smithy +++ b/smithy/models/main.smithy @@ -29,6 +29,7 @@ service Superposition { Variable Secret MasterKey + Job ] errors: [ InternalServerError diff --git a/smithy/smithy-build.json b/smithy/smithy-build.json index 0fed00e2a..c8c1928a9 100644 --- a/smithy/smithy-build.json +++ b/smithy/smithy-build.json @@ -80,7 +80,8 @@ "Workspace Management", "Secrets", "Audit & Monitoring", - "Variables" + "Variables", + "Background Jobs" ], "defaultBlobFormat": "byte", "keepUnusedComponents": false, From 51856f5e7370002248b37c428cf14add9aa479b2 Mon Sep 17 00:00:00 2001 From: datron Date: Tue, 21 Jul 2026 14:06:08 +0530 Subject: [PATCH 07/12] fix: rebase with main Signed-off-by: datron --- crates/context_aware_config/src/api/context/handlers.rs | 2 +- crates/service_utils/src/middlewares/auth_n.rs | 3 ++- crates/superposition/src/dispatch/handlers.rs | 9 +++++++-- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/crates/context_aware_config/src/api/context/handlers.rs b/crates/context_aware_config/src/api/context/handlers.rs index f66379e25..036133c59 100644 --- a/crates/context_aware_config/src/api/context/handlers.rs +++ b/crates/context_aware_config/src/api/context/handlers.rs @@ -1265,7 +1265,7 @@ async fn weight_recompute_handler( db_conn: DbConnection, ) -> superposition::Result> { let DbConnection(mut conn) = db_conn; - let job_request = JobRequest::PriorityRecompute(PriorityRecomputeRequest::default()); + let job_request = JobRequest::PriorityRecompute(PriorityRecomputeRequest); let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); let target_workspace = state .kronos_workspace diff --git a/crates/service_utils/src/middlewares/auth_n.rs b/crates/service_utils/src/middlewares/auth_n.rs index 7f8f2ebfc..7fd71ede8 100644 --- a/crates/service_utils/src/middlewares/auth_n.rs +++ b/crates/service_utils/src/middlewares/auth_n.rs @@ -113,7 +113,8 @@ fn process_basic_auth<'a>( } })?; - if request.path().ends_with("/dispatch/webhook") + if (request.path().ends_with("/dispatch/webhook") + || request.path().ends_with("/dispatch/job")) && is_dispatch_credential(&id, &secret, &state.kronos_dispatch_token) { let user = process_dipatcher_token(request); diff --git a/crates/superposition/src/dispatch/handlers.rs b/crates/superposition/src/dispatch/handlers.rs index 52ffd1adb..890b34fa6 100644 --- a/crates/superposition/src/dispatch/handlers.rs +++ b/crates/superposition/src/dispatch/handlers.rs @@ -116,8 +116,13 @@ async fn dispatch_job_handler( log::error!("Job {job_id} failed: {error_msg}"); update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; - append_job_logs(&mut conn, job_id, &format!("Job failed: {error_msg}"), Some(key)) - .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; + append_job_logs( + &mut conn, + job_id, + &format!("Job failed: {error_msg}"), + Some(key), + ) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; Err(unexpected_error!("Job {job_id} failed: {error_msg}")) } } From cbe3fbe7bc12d467e8da6beccd0d1d350ea20419 Mon Sep 17 00:00:00 2001 From: datron Date: Wed, 22 Jul 2026 12:45:40 +0530 Subject: [PATCH 08/12] fix: AI comments Signed-off-by: datron --- .../src/api/config/handlers.rs | 4 +- crates/service_utils/src/kronos_dispatch.rs | 103 ++++++++++++++---- crates/superposition/src/jobs/handlers.rs | 36 +++++- smithy/models/jobs.smithy | 37 +------ 4 files changed, 120 insertions(+), 60 deletions(-) diff --git a/crates/context_aware_config/src/api/config/handlers.rs b/crates/context_aware_config/src/api/config/handlers.rs index 28492c162..1eb48ade2 100644 --- a/crates/context_aware_config/src/api/config/handlers.rs +++ b/crates/context_aware_config/src/api/config/handlers.rs @@ -488,9 +488,11 @@ async fn reduce_handler( workspace_context: WorkspaceContext, db_conn: DbConnection, state: Data, + req: Json, ) -> superposition::Result> { let DbConnection(mut conn) = db_conn; - let job_request = JobRequest::Reduce(ReduceRequest::default()); + let req = req.into_inner(); + let job_request = JobRequest::Reduce(req); let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); let target_workspace = state .kronos_workspace diff --git a/crates/service_utils/src/kronos_dispatch.rs b/crates/service_utils/src/kronos_dispatch.rs index b917911af..a9c2ce9b2 100644 --- a/crates/service_utils/src/kronos_dispatch.rs +++ b/crates/service_utils/src/kronos_dispatch.rs @@ -14,7 +14,8 @@ use serde_json::json; use snowflake::SnowflakeIdGenerator; use superposition_types::{ DBConnection, - api::jobs::{JobCreateResponse, JobRequest}, + api::jobs::{JobCreateResponse, JobListFilters, JobRequest}, + custom_query::PaginationParams, database::{ models::{ BackgroundJob, BackgroundJobStatus, JobWorkspace, others::WorkspaceJobView, @@ -24,7 +25,7 @@ use superposition_types::{ }, }; -use crate::helpers::get_from_env_or_default; +use crate::{helpers::get_from_env_or_default, service::types::SchemaName}; static CONFIG_REFERENCE_REGEX: Lazy = Lazy::new(|| { regex::Regex::new(r"\{\{(?PVARS|SECRETS)\.(?P[A-Z0-9_]+)\}\}") @@ -377,27 +378,89 @@ pub fn get_job_by_id( pub fn list_jobs( conn: &mut DBConnection, - workspace: &JobWorkspace, - job_type: Option, - status: Option, -) -> anyhow::Result> { - let schema = workspace.as_db_string(); - let mut query = job_manager_view_dsl::job_manager - .schema_name(&schema) - .select(WorkspaceJobView::as_select()) - .into_boxed(); + schema: &SchemaName, + filters: &JobListFilters, + pagination: &PaginationParams, +) -> anyhow::Result<(i64, Vec)> { + let build_query = |f: &JobListFilters| { + let mut query = job_manager_view_dsl::job_manager + .schema_name(schema) + .into_boxed(); + if let Some(jt) = f.job_type { + query = query.filter(job_manager_view_dsl::job_type.eq(jt)); + } + if let Some(st) = f.status { + query = query.filter(job_manager_view_dsl::status.eq(st)); + } + query + }; - if let Some(jt) = job_type { - query = query.filter(job_manager_view_dsl::job_type.eq(jt)); - } - if let Some(st) = status { - query = query.filter(job_manager_view_dsl::status.eq(st)); + let base_query = build_query(filters); + + if let Some(true) = pagination.all { + let data = base_query + .order(job_manager_view_dsl::created_at.desc()) + .get_results(conn)?; + return Ok((data.len() as i64, data)); } - query + let count_query = build_query(filters); + + let total_items: i64 = count_query.count().get_result(conn)?; + + let limit = pagination.count.unwrap_or(10); + let mut paged_query = base_query .order(job_manager_view_dsl::created_at.desc()) - .load::(conn) - .map_err(|e| anyhow::anyhow!("Failed to list jobs: {e}")) + .limit(limit); + if let Some(page) = pagination.page { + let offset = (page - 1) * limit; + paged_query = paged_query.offset(offset); + } + let data = paged_query.get_results(conn)?; + + Ok((total_items, data)) +} + +pub fn list_jobs_global( + conn: &mut DBConnection, + filters: &JobListFilters, + pagination: &PaginationParams, +) -> anyhow::Result<(i64, Vec)> { + let build_query = |f: &JobListFilters| { + let mut query = job_manager_dsl::job_manager.into_boxed(); + if let Some(jt) = f.job_type { + query = query.filter(job_manager_dsl::job_type.eq(jt)); + } + if let Some(st) = f.status { + query = query.filter(job_manager_dsl::status.eq(st)); + } + query + }; + + let base_query = build_query(filters); + + if let Some(true) = pagination.all { + let data = base_query + .order(job_manager_dsl::created_at.desc()) + .get_results(conn)?; + return Ok((data.len() as i64, data)); + } + + let count_query = build_query(filters); + + let total_items: i64 = count_query.count().get_result(conn)?; + + let limit = pagination.count.unwrap_or(10); + let mut paged_query = base_query + .order(job_manager_dsl::created_at.desc()) + .limit(limit); + if let Some(page) = pagination.page { + let offset = (page - 1) * limit; + paged_query = paged_query.offset(offset); + } + let data = paged_query.get_results(conn)?; + + Ok((total_items, data)) } pub fn update_job_status( @@ -422,7 +485,7 @@ pub fn update_job_progress( .select(job_manager_dsl::progress) .first::(conn)?; - if (progress - previous_progress).abs() > threshold { + if progress == 100 || (progress - previous_progress).abs() > threshold { diesel::update( job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id)), ) diff --git a/crates/superposition/src/jobs/handlers.rs b/crates/superposition/src/jobs/handlers.rs index 1a6f6405d..a6fd6f320 100644 --- a/crates/superposition/src/jobs/handlers.rs +++ b/crates/superposition/src/jobs/handlers.rs @@ -9,8 +9,10 @@ use service_utils::{ use superposition_derives::{authorized, declare_resource}; use superposition_macros::unexpected_error; use superposition_types::{ + PaginatedResponse, api::jobs::{ExecutionDetails, JobDetailResponse, JobListFilters, JobResponse}, - database::models::{BackgroundJobStatus, JobWorkspace, others::WorkspaceJobView}, + custom_query::PaginationParams, + database::models::{BackgroundJobStatus, JobWorkspace}, result as superposition, }; @@ -28,16 +30,38 @@ pub fn endpoints() -> Scope { async fn list_handler( workspace_context: WorkspaceContext, db_conn: DbConnection, + pagination: Query, filters: Query, -) -> superposition::Result>> { +) -> superposition::Result>> { let DbConnection(mut conn) = db_conn; let filters = filters.into_inner(); - let job_workspace = JobWorkspace::from(&workspace_context.schema_name.0); + let pagination = pagination.into_inner(); + + let (total_items, jobs) = list_jobs( + &mut conn, + &workspace_context.schema_name, + &filters, + &pagination, + ) + .map_err(|e| unexpected_error!("Failed to list jobs: {}", e))?; + + let data: Vec = jobs + .into_iter() + .map(|j| JobResponse::from_view(&j, &workspace_context.schema_name.0)) + .collect(); + + if let Some(true) = pagination.all { + return Ok(Json(PaginatedResponse::all(data))); + } - let jobs = list_jobs(&mut conn, &job_workspace, filters.job_type, filters.status) - .map_err(|e| unexpected_error!("Failed to list jobs: {}", e))?; + let limit = pagination.count.unwrap_or(10); + let total_pages = (total_items as f64 / limit as f64).ceil() as i64; - Ok(Json(jobs)) + Ok(Json(PaginatedResponse { + total_pages, + total_items, + data, + })) } #[authorized] diff --git a/smithy/models/jobs.smithy b/smithy/models/jobs.smithy index 6796503ff..0be7d5583 100644 --- a/smithy/models/jobs.smithy +++ b/smithy/models/jobs.smithy @@ -11,7 +11,7 @@ resource Job { properties: { kronos_job_id: String description: String - type: BackgroundJobType + job_type: BackgroundJobType status: BackgroundJobStatus name: String progress: Integer @@ -92,38 +92,9 @@ structure JobDetailResponse for Job { execution: ExecutionDetails } -@documentation("Job summary returned by list operations. Does not include workspace_schema.") -structure JobSummary for Job { - @required - $id - - @required - $kronos_job_id - - @required - $description - - @required - $type - - @required - $status - - @required - $name - - @required - $progress - - @required - $created_at - - @required - $logs -} list JobList { - member: JobSummary + member: JobDetailResponse } @documentation("Response returned when a job is submitted. Contains the BJM job ID, Kronos job ID, and initial status.") @@ -143,7 +114,7 @@ structure JobCreateResponse for Job { @http(method: "GET", uri: "/jobs") @tags(["Background Jobs"]) operation ListJobs { - input := with [WorkspaceMixin] { + input := with [PaginationParams, WorkspaceMixin] { @httpQuery("status") @notProperty status: BackgroundJobStatus @@ -153,7 +124,7 @@ operation ListJobs { job_type: BackgroundJobType } - output := { + output := with [PaginatedResponse] { @required data: JobList } From 7e6d82d407d4d28c81a47899b1db0548e253f31b Mon Sep 17 00:00:00 2001 From: datron Date: Wed, 29 Jul 2026 17:54:49 +0530 Subject: [PATCH 09/12] feat: rebase with main Signed-off-by: datron --- .../src/api/config/handlers.rs | 16 ++++----- .../src/api/context/handlers.rs | 21 +++++------- crates/superposition/src/dispatch/handlers.rs | 34 ++++++++----------- 3 files changed, 30 insertions(+), 41 deletions(-) diff --git a/crates/context_aware_config/src/api/config/handlers.rs b/crates/context_aware_config/src/api/config/handlers.rs index 1eb48ade2..3fe6199c2 100644 --- a/crates/context_aware_config/src/api/config/handlers.rs +++ b/crates/context_aware_config/src/api/config/handlers.rs @@ -5,16 +5,17 @@ use actix_web::{ web::{Data, Header, Json, Path, Query}, }; use chrono::{DateTime, Utc}; -use diesel::{ExpressionMethods, QueryDsl, RunQueryDsl, SelectableHelper}; +use diesel::{ + ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl, SelectableHelper, + r2d2::{ConnectionManager, PooledConnection}, +}; use itertools::Itertools; use serde_json::{Map, Value, json}; use service_utils::{ helpers::{fetch_dimensions_info_map, is_not_modified}, kronos_dispatch::submit_job, redis::{CONFIG_KEY_SUFFIX, LAST_MODIFIED_KEY_SUFFIX, read_through_cache}, - service::types::{ - AppHeader, AppState, DbConnection, WorkspaceContext, WorkspaceWritePermit, - }, + service::types::{AppHeader, AppState, DbConnection, WorkspaceContext}, }; use superposition_core::{ ConfigFormat, JsonFormat, TomlFormat, @@ -448,12 +449,11 @@ async fn reduce_config_key( pub async fn execute_reduce( workspace_context: &WorkspaceContext, - mut write_permit: WorkspaceWritePermit, - user: &User, state: &Data, + conn: &mut PooledConnection>, + user: &User, is_approve: bool, ) -> superposition::Result<()> { - let conn = write_permit.connection(); let dimensions_info_map = fetch_dimensions_info_map(conn, &workspace_context.schema_name)?; let mut config = generate_cac(conn, &workspace_context.schema_name)?; @@ -463,7 +463,7 @@ pub async fn execute_reduce( let overrides = config.overrides; let default_config = config.default_configs.into_inner(); config = reduce_config_key( - &user, + user, conn, contexts.clone(), overrides.clone(), diff --git a/crates/context_aware_config/src/api/context/handlers.rs b/crates/context_aware_config/src/api/context/handlers.rs index 036133c59..782de754b 100644 --- a/crates/context_aware_config/src/api/context/handlers.rs +++ b/crates/context_aware_config/src/api/context/handlers.rs @@ -7,9 +7,10 @@ use actix_web::{ use bigdecimal::BigDecimal; use chrono::Utc; use diesel::{ - Connection, ExpressionMethods, OptionalExtension, QueryDsl, RunQueryDsl, - SelectableHelper, + Connection, ExpressionMethods, OptionalExtension, PgConnection, QueryDsl, + RunQueryDsl, SelectableHelper, dsl::sql, + r2d2::{ConnectionManager, PooledConnection}, sql_types::{Bool, Text}, }; use serde_json::{Map, Value}; @@ -29,7 +30,7 @@ use superposition_derives::{authorized, declare_resource}; use superposition_macros::{bad_argument, db_error, unexpected_error}; use superposition_types::{ Contextual, DBConnection, DimensionInfo, InternalUserContext, Overridden, Overrides, - PaginatedResponse, Resource, SortBy, User, + PaginatedResponse, PrefixList, Resource, SortBy, User, api::{ DimensionMatchStrategy, context::{ @@ -1168,15 +1169,13 @@ async fn bulk_operations_handler( pub async fn execute_priority_recompute( workspace_context: &WorkspaceContext, state: &Data, - mut write_permit: WorkspaceWritePermit, + conn: &mut PooledConnection>, user: &User, ) -> superposition::Result<()> { use superposition_types::database::schema::contexts::dsl::{ contexts, last_modified_at, last_modified_by, weight, }; - let conn = write_permit.connection(); - let result: Vec = contexts .schema_name(&workspace_context.schema_name) .load(conn) @@ -1237,13 +1236,9 @@ pub async fn execute_priority_recompute( .map_err(|e| unexpected_error!(e))?; add_config_version(state, None, config_version_desc, transaction_conn, &workspace_context.schema_name) })?; - let _ = put_config_in_redis( - &config_version, - &state, - &workspace_context.schema_name, - conn, - ) - .await; + let _ = + put_config_in_redis(&config_version, state, &workspace_context.schema_name, conn) + .await; let data = WebhookData { payload: &response, diff --git a/crates/superposition/src/dispatch/handlers.rs b/crates/superposition/src/dispatch/handlers.rs index 890b34fa6..4f4e7f819 100644 --- a/crates/superposition/src/dispatch/handlers.rs +++ b/crates/superposition/src/dispatch/handlers.rs @@ -17,7 +17,7 @@ use service_utils::{ append_job_logs, has_pattern_in_headers, substitute_templates, update_job_progress, update_job_status, }, - service::types::{AppState, DbConnection, WorkspaceContext}, + service::types::{AppState, DbConnection, WorkspaceContext, WorkspaceWritePermit}, }; use superposition_derives::{authorized, declare_resource}; use superposition_macros::unexpected_error; @@ -66,7 +66,7 @@ async fn dispatch_handler( async fn dispatch_job_handler( workspace_context: WorkspaceContext, state: Data, - db_conn: DbConnection, + mut write_permit: WorkspaceWritePermit, user: User, body: Json, ) -> superposition::Result { @@ -74,11 +74,11 @@ async fn dispatch_job_handler( job_id, job_request, } = body.into_inner(); - let DbConnection(mut conn) = db_conn; - update_job_status(&mut conn, job_id, BackgroundJobStatus::Inprogress) + let conn = write_permit.connection(); + update_job_status(conn, job_id, BackgroundJobStatus::Inprogress) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; - let key = append_job_logs(&mut conn, job_id, "Job started", None) + let key = append_job_logs(conn, job_id, "Job started", None) .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; let result = match &job_request { @@ -86,43 +86,37 @@ async fn dispatch_job_handler( execute_webhook_dispatch( &workspace_context, &state, - &mut conn, + conn, &req.webhook_name, &req.data, ) .await } JobRequest::PriorityRecompute(_) => { - execute_priority_recompute(&workspace_context, &state, &mut conn, &user).await + execute_priority_recompute(&workspace_context, &state, conn, &user).await } JobRequest::Reduce(req) => { - execute_reduce(&workspace_context, &state, &mut conn, &user, req.approve) - .await + execute_reduce(&workspace_context, &state, conn, &user, req.approve).await } }; match result { Ok(()) => { - update_job_status(&mut conn, job_id, BackgroundJobStatus::Completed) + update_job_status(conn, job_id, BackgroundJobStatus::Completed) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; - update_job_progress(&mut conn, job_id, 100) + update_job_progress(conn, job_id, 100) .map_err(|e| unexpected_error!("Failed to update job progress: {}", e))?; - append_job_logs(&mut conn, job_id, "Job completed", Some(key)) + append_job_logs(conn, job_id, "Job completed", Some(key)) .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; Ok(HttpResponse::Ok().finish()) } Err(e) => { let error_msg = format!("{e}"); log::error!("Job {job_id} failed: {error_msg}"); - update_job_status(&mut conn, job_id, BackgroundJobStatus::Failed) + update_job_status(conn, job_id, BackgroundJobStatus::Failed) .map_err(|e| unexpected_error!("Failed to update job status: {}", e))?; - append_job_logs( - &mut conn, - job_id, - &format!("Job failed: {error_msg}"), - Some(key), - ) - .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; + append_job_logs(conn, job_id, &format!("Job failed: {error_msg}"), Some(key)) + .map_err(|e| unexpected_error!("Failed to append logs: {}", e))?; Err(unexpected_error!("Job {job_id} failed: {error_msg}")) } } From d5a2139c2df3c2e0a78cddd97c8bc30eca99a463 Mon Sep 17 00:00:00 2001 From: datron Date: Tue, 4 Aug 2026 15:24:19 +0530 Subject: [PATCH 10/12] fix: create KronosJobRequest struct Signed-off-by: datron --- .../src/api/config/handlers.rs | 2 +- .../src/api/context/handlers.rs | 2 +- crates/service_utils/src/kronos_dispatch.rs | 19 +++++++++---------- crates/superposition_types/src/api/jobs.rs | 10 ++++++++++ 4 files changed, 21 insertions(+), 12 deletions(-) diff --git a/crates/context_aware_config/src/api/config/handlers.rs b/crates/context_aware_config/src/api/config/handlers.rs index 3fe6199c2..88c494cff 100644 --- a/crates/context_aware_config/src/api/config/handlers.rs +++ b/crates/context_aware_config/src/api/config/handlers.rs @@ -505,7 +505,7 @@ async fn reduce_handler( &job_workspace, &workspace_context.organisation_id, &workspace_context.workspace_id, - &job_request, + job_request, &state.snowflake_generator, &mut conn, 3, diff --git a/crates/context_aware_config/src/api/context/handlers.rs b/crates/context_aware_config/src/api/context/handlers.rs index 782de754b..43cd904af 100644 --- a/crates/context_aware_config/src/api/context/handlers.rs +++ b/crates/context_aware_config/src/api/context/handlers.rs @@ -1273,7 +1273,7 @@ async fn weight_recompute_handler( &job_workspace, &workspace_context.organisation_id, &workspace_context.workspace_id, - &job_request, + job_request, &state.snowflake_generator, &mut conn, 3, diff --git a/crates/service_utils/src/kronos_dispatch.rs b/crates/service_utils/src/kronos_dispatch.rs index a9c2ce9b2..ad66b5764 100644 --- a/crates/service_utils/src/kronos_dispatch.rs +++ b/crates/service_utils/src/kronos_dispatch.rs @@ -14,7 +14,7 @@ use serde_json::json; use snowflake::SnowflakeIdGenerator; use superposition_types::{ DBConnection, - api::jobs::{JobCreateResponse, JobListFilters, JobRequest}, + api::jobs::{JobCreateResponse, JobListFilters, JobRequest, KronosJobRequest}, custom_query::PaginationParams, database::{ models::{ @@ -263,7 +263,7 @@ pub async fn submit_job( workspace: &JobWorkspace, org_id: &str, workspace_id: &str, - job_request: &JobRequest, + job_request: JobRequest, snowflake_generator: &Arc>, conn: &mut DBConnection, max_attempts: i64, @@ -297,13 +297,12 @@ pub async fn submit_job( .values(&bjm_entry) .execute(conn)?; - let job_request_value = serde_json::to_value(job_request)?; - let mut input = job_request_value; - if let Some(obj) = input.as_object_mut() { - obj.insert("org_id".to_string(), json!(org_id)); - obj.insert("workspace".to_string(), json!(workspace_id)); - obj.insert("job_id".to_string(), json!(job_id.to_string())); - } + let input = KronosJobRequest { + request: job_request, + org_id: org_id.to_string(), + workspace_id: workspace_id.to_string(), + job_id, + }; let idempotency_key = format!( "{}_{}_{}_{}", @@ -317,7 +316,7 @@ pub async fn submit_job( .create_job( target_workspace, JOB_DISPATCHER_ENDPOINT_NAME, - input, + json!(input), max_attempts, JobTrigger::Immediate, Some(&idempotency_key), diff --git a/crates/superposition_types/src/api/jobs.rs b/crates/superposition_types/src/api/jobs.rs index b08208913..b56eafa96 100644 --- a/crates/superposition_types/src/api/jobs.rs +++ b/crates/superposition_types/src/api/jobs.rs @@ -88,6 +88,16 @@ impl JobResponse { } } +#[derive(Debug, Clone, Serialize)] +pub struct KronosJobRequest { + #[serde(flatten)] + pub request: JobRequest, + pub org_id: String, + pub workspace_id: String, + #[serde(with = "crate::database::models::i64_formatter")] + pub job_id: i64, +} + impl From for JobResponse { fn from(job: BackgroundJob) -> Self { Self { From 1f92ebd22d1117c5cb05b83d88512177c6ee18d5 Mon Sep 17 00:00:00 2001 From: datron Date: Mon, 10 Aug 2026 17:06:21 +0530 Subject: [PATCH 11/12] fix: append logs Signed-off-by: datron --- crates/service_utils/src/kronos_dispatch.rs | 14 ++++++++++---- crates/superposition_types/src/api/jobs.rs | 1 + 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/crates/service_utils/src/kronos_dispatch.rs b/crates/service_utils/src/kronos_dispatch.rs index ad66b5764..c1c8b8866 100644 --- a/crates/service_utils/src/kronos_dispatch.rs +++ b/crates/service_utils/src/kronos_dispatch.rs @@ -508,10 +508,16 @@ pub fn append_job_logs( let timed_key = key.unwrap_or(Utc::now().to_rfc3339()); let mut logs = current.as_object().cloned().unwrap_or_default(); - logs.insert( - timed_key.clone(), - serde_json::Value::String(log_line.to_string()), - ); + let final_log_lines = match logs.get(&timed_key) { + Some(serde_json::Value::String(log_lines)) => { + let mut updated_lines = log_lines.clone(); + updated_lines.push_str("\n"); + updated_lines.push_str(log_line); + serde_json::Value::String(updated_lines) + } + _ => serde_json::Value::String(log_line.to_string()), + }; + logs.insert(timed_key.clone(), final_log_lines); let new_logs = serde_json::Value::Object(logs); diesel::update(job_manager_dsl::job_manager.filter(job_manager_dsl::id.eq(job_id))) diff --git a/crates/superposition_types/src/api/jobs.rs b/crates/superposition_types/src/api/jobs.rs index b56eafa96..2a04d56f6 100644 --- a/crates/superposition_types/src/api/jobs.rs +++ b/crates/superposition_types/src/api/jobs.rs @@ -93,6 +93,7 @@ pub struct KronosJobRequest { #[serde(flatten)] pub request: JobRequest, pub org_id: String, + #[serde(rename = "workspace")] pub workspace_id: String, #[serde(with = "crate::database::models::i64_formatter")] pub job_id: i64, From 90eb431b7c9dbbc3500507982bd68b4c176a945c Mon Sep 17 00:00:00 2001 From: datron Date: Tue, 18 Aug 2026 11:18:17 +0530 Subject: [PATCH 12/12] fix: resolve comments Signed-off-by: datron --- crates/service_utils/src/kronos_dispatch.rs | 2 +- .../2026-07-16-000001_job_manager/down.sql | 12 ++----- .../2026-07-16-000001_job_manager/up.sql | 34 +++++++------------ 3 files changed, 16 insertions(+), 32 deletions(-) diff --git a/crates/service_utils/src/kronos_dispatch.rs b/crates/service_utils/src/kronos_dispatch.rs index c1c8b8866..9503df434 100644 --- a/crates/service_utils/src/kronos_dispatch.rs +++ b/crates/service_utils/src/kronos_dispatch.rs @@ -511,7 +511,7 @@ pub fn append_job_logs( let final_log_lines = match logs.get(&timed_key) { Some(serde_json::Value::String(log_lines)) => { let mut updated_lines = log_lines.clone(); - updated_lines.push_str("\n"); + updated_lines.push('\n'); updated_lines.push_str(log_line); serde_json::Value::String(updated_lines) } diff --git a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql index c65b7e635..7d575e213 100644 --- a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql +++ b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/down.sql @@ -7,14 +7,6 @@ DROP INDEX IF EXISTS superposition.idx_job_manager_kronos_job_id; DROP INDEX IF EXISTS superposition.idx_job_manager_status_job_type; DROP INDEX IF EXISTS superposition.idx_job_manager_created_at; -DO $$ BEGIN - DROP TYPE public.background_job_type; -EXCEPTION - WHEN undefined_object THEN null; -END $$; +DROP TYPE public.background_job_type; -DO $$ BEGIN - DROP TYPE public.background_job_status; -EXCEPTION - WHEN undefined_object THEN null; -END $$; +DROP TYPE public.background_job_status; diff --git a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql index c905367e6..1f8b0ded1 100644 --- a/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql +++ b/crates/superposition_types/migrations/2026-07-16-000001_job_manager/up.sql @@ -1,24 +1,16 @@ -DO $$ BEGIN - CREATE TYPE public.background_job_type AS ENUM ( - 'WEBHOOK', - 'PRIORITY_RECOMPUTE', - 'REDUCE' - ); -EXCEPTION - WHEN duplicate_object THEN null; -END $$; - -DO $$ BEGIN - CREATE TYPE public.background_job_status AS ENUM ( - 'CREATED', - 'SCHEDULED', - 'INPROGRESS', - 'FAILED', - 'COMPLETED' - ); -EXCEPTION - WHEN duplicate_object THEN null; -END $$; +CREATE TYPE public.background_job_type AS ENUM ( + 'WEBHOOK', + 'PRIORITY_RECOMPUTE', + 'REDUCE' +); + +CREATE TYPE public.background_job_status AS ENUM ( + 'CREATED', + 'SCHEDULED', + 'INPROGRESS', + 'FAILED', + 'COMPLETED' +); CREATE TABLE IF NOT EXISTS superposition.job_manager ( id BIGINT PRIMARY KEY,