pharos_runtime_database/

queue.rs

//! Bounded queue readiness reads over the configured database adapter.

use serde_json::Value;
use std::time::{SystemTime, UNIX_EPOCH};

use crate::DbBackendConfig;
use crate::operations::{
    Aggregate, AggregateQuery, AppSchema, DatabaseError, DatabaseOperations, Mutation, Predicate,
    Query, SortKey, Transaction, WriteValue,
};
use std::collections::BTreeMap;

/// Applies a declared queue control with a compare-and-write transition.
/// Admission and role policy belong to the caller; active actor state and the
/// expected job status are checked atomically with the mutation.
pub fn control_job(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    actor_id: i64,
    execution_id: i64,
    retry: bool,
    now: &str,
) -> Result<(), DatabaseError> {
    if actor_id <= 0 || execution_id <= 0 {
        return Err(DatabaseError::InvalidInput);
    }
    let statuses = if retry {
        vec![
            Value::String("failed".to_owned()),
            Value::String("cancelled".to_owned()),
        ]
    } else {
        vec![
            Value::String("queued".to_owned()),
            Value::String("running".to_owned()),
        ]
    };
    let current = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: vec!["id".to_owned(), "execution_status".to_owned()],
            filter: Some(Predicate::Equal("id".to_owned(), Value::from(execution_id))),
            order: Vec::new(),
            limit: 1,
            offset: 0,
        },
    )?;
    let row = current.first().ok_or(DatabaseError::MissingRecord)?;
    let status = row
        .get("execution_status")
        .cloned()
        .ok_or(DatabaseError::StorageFailure)?;
    if !statuses.contains(&status) {
        return Err(DatabaseError::Conflict);
    }
    let mut values = BTreeMap::from([
        (
            "execution_status".to_owned(),
            WriteValue::Literal(Value::String(
                if retry { "queued" } else { "cancelled" }.to_owned(),
            )),
        ),
        (
            "lease_expires_at".to_owned(),
            WriteValue::Literal(Value::Null),
        ),
    ]);
    if retry {
        values.insert(
            "available_at".to_owned(),
            WriteValue::Literal(Value::String(now.to_owned())),
        );
        values.insert("claimed_at".to_owned(), WriteValue::Literal(Value::Null));
        values.insert("failed_at".to_owned(), WriteValue::Literal(Value::Null));
        values.insert("completed_at".to_owned(), WriteValue::Literal(Value::Null));
        values.insert(
            "attempt_count".to_owned(),
            WriteValue::Literal(Value::from(0)),
        );
        values.insert(
            "last_error".to_owned(),
            WriteValue::Literal(Value::String(String::new())),
        );
    }
    adapter.transact(
        config,
        schema,
        &Transaction {
            steps: vec![
                Mutation::RequireCount {
                    entity: "user_account".to_owned(),
                    filter: Predicate::All(vec![
                        Predicate::Equal("id".to_owned(), Value::from(actor_id)),
                        Predicate::Equal("is_active".to_owned(), Value::Bool(true)),
                    ]),
                    minimum: 1,
                    maximum: Some(1),
                },
                Mutation::Update {
                    entity: "job_execution".to_owned(),
                    filter: Predicate::All(vec![
                        Predicate::Equal("id".to_owned(), Value::from(execution_id)),
                        Predicate::Equal("execution_status".to_owned(), status),
                    ]),
                    values,
                    expected_rows: Some(1),
                },
            ],
            idempotency_key: None,
        },
    )?;
    Ok(())
}

/// Reads one execution and its bounded attempts for declared detail rendering.
pub fn job_detail_view(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    execution_id: i64,
) -> Result<Value, DatabaseError> {
    if execution_id <= 0 {
        return Err(DatabaseError::InvalidInput);
    }
    let fields = |name: &str| -> Result<Vec<String>, DatabaseError> {
        Ok(schema
            .entities
            .iter()
            .find(|entity| entity.name == name)
            .ok_or(DatabaseError::InvalidContract)?
            .fields
            .iter()
            .map(|field| field.name.clone())
            .collect())
    };
    let executions = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: fields("job_execution")?,
            filter: Some(Predicate::Equal("id".to_owned(), Value::from(execution_id))),
            order: Vec::new(),
            limit: 1,
            offset: 0,
        },
    )?;
    if executions.is_empty() {
        return Err(DatabaseError::MissingRecord);
    }
    let attempts = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution_attempt".to_owned(),
            fields: fields("job_execution_attempt")?,
            filter: Some(Predicate::Equal(
                "job_execution_id".to_owned(),
                Value::from(execution_id),
            )),
            order: vec![SortKey {
                field: "id".to_owned(),
                descending: false,
            }],
            limit: 257,
            offset: 0,
        },
    )?;
    if attempts.len() > 256 {
        return Err(DatabaseError::CapacityExceeded);
    }
    Ok(serde_json::json!({"job_executions": executions, "job_execution_attempts": attempts}))
}

/// Reads one queue page and aggregate counts without exporting unrelated app
/// state. The declared route policy authorizes queue access before this call.
pub fn job_queue_page(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    status: &str,
    search: &str,
    requested_page: usize,
) -> Result<Value, DatabaseError> {
    if search.chars().count() > 160 || requested_page == 0 || requested_page > 50_000 {
        return Err(DatabaseError::InvalidInput);
    }
    let mut filters = Vec::new();
    if !status.is_empty() {
        if !matches!(
            status,
            "queued" | "running" | "succeeded" | "failed" | "cancelled"
        ) {
            return Err(DatabaseError::InvalidInput);
        }
        filters.push(Predicate::Equal(
            "execution_status".to_owned(),
            Value::String(status.to_owned()),
        ));
    }
    if !search.is_empty() {
        filters.push(Predicate::Any(vec![
            Predicate::Contains("job_id".to_owned(), search.to_owned()),
            Predicate::Contains("handler_id".to_owned(), search.to_owned()),
        ]));
    }
    let filter = (!filters.is_empty()).then_some(Predicate::All(filters));
    let count = adapter
        .aggregate(
            config,
            schema,
            &AggregateQuery {
                entity: "job_execution".to_owned(),
                filter: filter.clone(),
                aggregate: Aggregate::Count,
            },
        )?
        .as_u64()
        .ok_or(DatabaseError::StorageFailure)?;
    let page_count = count.div_ceil(20).max(1);
    let page = (requested_page as u64).min(page_count);
    let definition = schema
        .entities
        .iter()
        .find(|entity| entity.name == "job_execution")
        .ok_or(DatabaseError::InvalidContract)?;
    let rows = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: definition
                .fields
                .iter()
                .map(|field| field.name.clone())
                .collect(),
            filter,
            order: vec![SortKey {
                field: "id".to_owned(),
                descending: false,
            }],
            limit: 20,
            offset: u32::try_from((page - 1) * 20).map_err(|_| DatabaseError::InvalidInput)?,
        },
    )?;
    let mut counts = serde_json::Map::new();
    for status in ["queued", "running", "succeeded", "failed", "cancelled"] {
        let count = adapter.aggregate(
            config,
            schema,
            &AggregateQuery {
                entity: "job_execution".to_owned(),
                filter: Some(Predicate::Equal(
                    "execution_status".to_owned(),
                    Value::String(status.to_owned()),
                )),
                aggregate: Aggregate::Count,
            },
        )?;
        counts.insert(status.to_owned(), count);
    }
    Ok(
        serde_json::json!({"job_executions": rows, "_job_page": {"counts": counts, "matched_count": count, "page": page}}),
    )
}

/// Checks for pending work without exporting the app state document.
/// A schema without the complete queue contract returns `None` so its caller
/// can retain its existing worker behavior during adoption.
pub fn pending_queue_work(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
) -> Result<Option<bool>, DatabaseError> {
    let families = [
        (
            "notification_delivery",
            "delivery_status",
            &["queued", "retryable"][..],
        ),
        (
            "integration_delivery",
            "delivery_status",
            &["queued", "retryable"][..],
        ),
        ("job_execution", "execution_status", &["queued"][..]),
    ];
    if families.iter().any(|(entity, _, _)| {
        !schema
            .entities
            .iter()
            .any(|declared| declared.name == *entity)
    }) {
        return Ok(None);
    }
    requeue_local_jobs(adapter, config, schema, false, 32)?;
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| DatabaseError::Unavailable)?
        .as_secs()
        .to_string();
    let now_utc = chrono::DateTime::<chrono::Utc>::from_timestamp(
        now.parse::<i64>()
            .map_err(|_| DatabaseError::InvalidInput)?,
        0,
    )
    .ok_or(DatabaseError::InvalidInput)?
    .to_rfc3339();
    for (entity, status_field, statuses) in families {
        let filter = if entity == "job_execution" {
            Predicate::All(vec![
                Predicate::Equal(status_field.to_owned(), Value::String("queued".to_owned())),
                Predicate::AtMost("available_at".to_owned(), Value::String(now_utc.clone())),
            ])
        } else if statuses.contains(&"retryable") {
            Predicate::Any(vec![
                Predicate::Equal(status_field.to_owned(), Value::String("queued".to_owned())),
                Predicate::All(vec![
                    Predicate::Equal(
                        status_field.to_owned(),
                        Value::String("retryable".to_owned()),
                    ),
                    Predicate::AtMost("next_attempt_at".to_owned(), Value::String(now.clone())),
                ]),
            ])
        } else {
            Predicate::In(
                status_field.to_owned(),
                statuses
                    .iter()
                    .map(|status| Value::String((*status).to_owned()))
                    .collect(),
            )
        };
        let rows = adapter.query(
            config,
            schema,
            &Query {
                entity: entity.to_owned(),
                fields: vec!["id".to_owned()],
                filter: Some(filter),
                order: Vec::new(),
                limit: 1,
                offset: 0,
            },
        )?;
        if !rows.is_empty() {
            return Ok(Some(true));
        }
    }
    Ok(Some(false))
}

/// Reads the next due records and related notification display data. Evidence
/// ID seeds contain no historical payloads or unrelated users.
pub fn scoped_job_worker_snapshot(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
) -> Result<Option<Value>, DatabaseError> {
    for name in [
        "notification_delivery",
        "integration_delivery",
        "job_execution",
        "job_execution_attempt",
        "workflow_execution_event",
        "integration_delivery_attempt",
    ] {
        if !schema.entities.iter().any(|entity| entity.name == name) {
            return Ok(None);
        }
    }
    let epoch = i64::try_from(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| DatabaseError::Unavailable)?
            .as_secs(),
    )
    .map_err(|_| DatabaseError::InvalidInput)?;
    let all_fields = |entity: &str| -> Result<Vec<String>, DatabaseError> {
        Ok(schema
            .entities
            .iter()
            .find(|definition| definition.name == entity)
            .ok_or(DatabaseError::InvalidContract)?
            .fields
            .iter()
            .map(|field| field.name.clone())
            .collect())
    };
    let mut deliveries = BTreeMap::new();
    for (entity, section) in [
        ("notification_delivery", "notification_deliveries"),
        ("integration_delivery", "integration_deliveries"),
    ] {
        let rows = adapter.query(
            config,
            schema,
            &Query {
                entity: entity.to_owned(),
                fields: all_fields(entity)?,
                filter: Some(Predicate::Any(vec![
                    Predicate::Equal(
                        "delivery_status".to_owned(),
                        Value::String("queued".to_owned()),
                    ),
                    Predicate::All(vec![
                        Predicate::Equal(
                            "delivery_status".to_owned(),
                            Value::String("retryable".to_owned()),
                        ),
                        Predicate::AtMost(
                            "next_attempt_at".to_owned(),
                            Value::String(epoch.to_string()),
                        ),
                    ]),
                ])),
                order: vec![SortKey {
                    field: "id".to_owned(),
                    descending: false,
                }],
                limit: 1,
                offset: 0,
            },
        )?;
        deliveries.insert(section, rows);
    }
    let due = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: schema
                .entities
                .iter()
                .find(|item| item.name == "job_execution")
                .ok_or(DatabaseError::InvalidContract)?
                .fields
                .iter()
                .map(|field| field.name.clone())
                .collect(),
            filter: Some(Predicate::All(vec![
                Predicate::Equal(
                    "execution_status".to_owned(),
                    Value::String("queued".to_owned()),
                ),
                Predicate::AtMost(
                    "available_at".to_owned(),
                    Value::String(
                        chrono::DateTime::<chrono::Utc>::from_timestamp(epoch, 0)
                            .ok_or(DatabaseError::InvalidInput)?
                            .to_rfc3339(),
                    ),
                ),
            ])),
            order: vec![SortKey {
                field: "id".to_owned(),
                descending: false,
            }],
            limit: 1,
            offset: 0,
        },
    )?;
    if due.is_empty() && deliveries.values().all(Vec::is_empty) {
        return Ok(None);
    }
    let latest = |entity_name: &str| -> Result<Vec<serde_json::Map<String, Value>>, DatabaseError> {
        let _entity = schema
            .entities
            .iter()
            .find(|item| item.name == entity_name)
            .ok_or(DatabaseError::InvalidContract)?;
        adapter.query(
            config,
            schema,
            &Query {
                entity: entity_name.to_owned(),
                fields: vec!["id".to_owned()],
                filter: None,
                order: vec![SortKey {
                    field: "id".to_owned(),
                    descending: true,
                }],
                limit: 1,
                offset: 0,
            },
        )
    };
    let mut notification = deliveries
        .remove("notification_deliveries")
        .ok_or(DatabaseError::InvalidContract)?;
    let mut integrations = deliveries
        .remove("integration_deliveries")
        .ok_or(DatabaseError::InvalidContract)?;
    for row in &mut integrations {
        if let Some(value) = row.get("signature_timestamp_header").cloned() {
            row.insert("timestamp_header".to_owned(), value);
        }
    }
    let mut users = Vec::new();
    let mut resets = Vec::new();
    let mut invites = Vec::new();
    let mut businesses = Vec::new();
    for row in &mut notification {
        let kind = row
            .get("template_id")
            .and_then(Value::as_str)
            .ok_or(DatabaseError::StorageFailure)?
            .to_owned();
        if matches!(kind.as_str(), "password_reset" | "business_invite") {
            let secret = row
                .get("resource_id")
                .and_then(Value::as_str)
                .ok_or(DatabaseError::StorageFailure)?
                .to_owned();
            let hash = crate::operations::bytes_fingerprint(secret.as_bytes());
            let entity = if kind == "password_reset" {
                "password_reset"
            } else {
                "invite"
            };
            let records = adapter.query(
                config,
                schema,
                &Query {
                    entity: entity.to_owned(),
                    fields: all_fields(entity)?,
                    filter: Some(Predicate::Equal(
                        "token_hash".to_owned(),
                        Value::String(hash),
                    )),
                    order: Vec::new(),
                    limit: 2,
                    offset: 0,
                },
            )?;
            if records.len() > 1 {
                return Err(DatabaseError::Conflict);
            }
            if kind == "password_reset" {
                let ids = records
                    .iter()
                    .filter_map(|row| row.get("user_id").cloned())
                    .collect::<Vec<_>>();
                users = crate::operations::query_related_rows(
                    adapter,
                    config,
                    schema,
                    crate::operations::RelatedRowsQuery {
                        entity: "user_account",
                        foreign_key: "id",
                        keys: &ids,
                        fields: &[
                            "id".to_owned(),
                            "display_name".to_owned(),
                            "username".to_owned(),
                            "email".to_owned(),
                        ],
                        maximum: 1,
                    },
                )?;
                resets = records;
            } else {
                for mut invite in records {
                    let tenant_id = invite
                        .get("target_tenant_id")
                        .cloned()
                        .ok_or(DatabaseError::StorageFailure)?;
                    let tenants = adapter.query(
                        config,
                        schema,
                        &Query {
                            entity: "tenant".to_owned(),
                            fields: vec!["source_business_id".to_owned()],
                            filter: Some(Predicate::Equal("id".to_owned(), tenant_id)),
                            order: Vec::new(),
                            limit: 1,
                            offset: 0,
                        },
                    )?;
                    let business_ids = tenants
                        .iter()
                        .filter_map(|row| row.get("source_business_id").cloned())
                        .collect::<Vec<_>>();
                    businesses = crate::operations::query_related_rows(
                        adapter,
                        config,
                        schema,
                        crate::operations::RelatedRowsQuery {
                            entity: "business",
                            foreign_key: "id",
                            keys: &business_ids,
                            fields: &["id".to_owned(), "name".to_owned()],
                            maximum: 1,
                        },
                    )?;
                    if let Some(id) = business_ids.first() {
                        invite.insert("business_id".to_owned(), id.clone());
                    }
                    invites.push(invite);
                }
            }
            row.insert("secret".to_owned(), Value::String(secret));
            row.insert("delivery_kind".to_owned(), Value::String(kind));
            if matches!(
                row.get("adapter_id").and_then(Value::as_str),
                Some("local_reset_queue" | "local_invite_queue")
            ) {
                row.insert(
                    "adapter_kind".to_owned(),
                    Value::String("notification_local".to_owned()),
                );
            }
        }
    }
    let mut snapshot = serde_json::json!({
        "_worker_scope": "queue",
        "job_executions": due,
        "job_execution_attempts": latest("job_execution_attempt")?,
        "workflow_execution_events": latest("workflow_execution_event")?,
        "notification_deliveries": notification,
        "integration_deliveries": integrations,
        "integration_delivery_attempts": latest("integration_delivery_attempt")?,
        "users": users, "password_resets": resets, "business_invites": invites, "businesses": businesses,
    });
    claim_worker_rows(adapter, config, schema, &mut snapshot, epoch)?;
    if snapshot
        .get("_worker_claims")
        .and_then(Value::as_array)
        .is_none_or(Vec::is_empty)
    {
        return Ok(None);
    }
    Ok(Some(snapshot))
}

/// Acquires adapter-owned leases before any delivery side effect. The transient
/// view retains the due status expected by the existing processors, while
/// completion guards compare the committed lease to prevent stale acknowledgments.
fn claim_worker_rows(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    snapshot: &mut Value,
    epoch: i64,
) -> Result<(), DatabaseError> {
    use crate::operations::Claim;
    let mut claims = Vec::new();
    for (section, entity, status, lease) in [
        (
            "notification_deliveries",
            "notification_delivery",
            "delivery_status",
            "next_attempt_at",
        ),
        (
            "integration_deliveries",
            "integration_delivery",
            "delivery_status",
            "next_attempt_at",
        ),
        (
            "job_executions",
            "job_execution",
            "execution_status",
            "lease_expires_at",
        ),
    ] {
        let rows = snapshot
            .get_mut(section)
            .and_then(Value::as_array_mut)
            .ok_or(DatabaseError::InvalidInput)?;
        let mut retained = Vec::new();
        for row in rows.drain(..) {
            let id = row
                .get("id")
                .and_then(Value::as_i64)
                .ok_or(DatabaseError::InvalidInput)?;
            let old_status = row
                .get(status)
                .cloned()
                .ok_or(DatabaseError::InvalidInput)?;
            let expires = epoch.checked_add(120).ok_or(DatabaseError::InvalidInput)?;
            let job = entity == "job_execution";
            let lease_value = if job {
                Value::String(
                    chrono::DateTime::<chrono::Utc>::from_timestamp(expires, 0)
                        .ok_or(DatabaseError::InvalidInput)?
                        .to_rfc3339(),
                )
            } else {
                Value::String(expires.to_string())
            };
            let claim_status = Value::String(if job { "running" } else { "retryable" }.to_owned());
            let mut filter = vec![
                Predicate::Equal("id".to_owned(), Value::from(id)),
                Predicate::Equal(status.to_owned(), old_status),
            ];
            if let Some(previous) = row.get(lease) {
                filter.push(if previous.is_null() {
                    Predicate::IsNull(lease.to_owned())
                } else {
                    Predicate::Equal(lease.to_owned(), previous.clone())
                });
            }
            let values = BTreeMap::from([
                (status.to_owned(), WriteValue::Literal(claim_status.clone())),
                (lease.to_owned(), WriteValue::Literal(lease_value.clone())),
            ]);
            if adapter.claim_one(
                config,
                schema,
                &Claim {
                    entity: entity.to_owned(),
                    filter: Predicate::All(filter),
                    order: vec![crate::operations::SortKey {
                        field: "id".to_owned(),
                        descending: false,
                    }],
                    values,
                },
            )? == Some(id)
            {
                claims.push(serde_json::json!({"entity": entity, "id": id, "fields": {status: claim_status, lease: lease_value}}));
                retained.push(row);
            }
        }
        *rows = retained;
    }
    snapshot
        .as_object_mut()
        .ok_or(DatabaseError::InvalidInput)?
        .insert("_worker_claims".to_owned(), Value::Array(claims));
    Ok(())
}

/// Marks one queued local job complete with a guarded, backend-neutral write.
/// This operator operation has no external handler side effects.
pub fn complete_one_local_job(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
) -> Result<Option<i64>, DatabaseError> {
    let rows = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: [
                "id",
                "attempt_count",
                "max_attempts",
                "workflow_execution_id",
                "job_id",
            ]
            .into_iter()
            .map(str::to_owned)
            .collect(),
            filter: Some(Predicate::All(vec![
                Predicate::Equal(
                    "execution_status".to_owned(),
                    Value::String("queued".to_owned()),
                ),
                Predicate::AtMost(
                    "available_at".to_owned(),
                    Value::String(chrono::Utc::now().to_rfc3339()),
                ),
            ])),
            order: vec![SortKey {
                field: "id".to_owned(),
                descending: false,
            }],
            limit: 1,
            offset: 0,
        },
    )?;
    let Some(row) = rows.first() else {
        return Ok(None);
    };
    let id = row
        .get("id")
        .and_then(Value::as_i64)
        .ok_or(DatabaseError::StorageFailure)?;
    let attempts = row
        .get("attempt_count")
        .and_then(Value::as_i64)
        .ok_or(DatabaseError::StorageFailure)?
        .checked_add(1)
        .ok_or(DatabaseError::InvalidInput)?;
    if attempts
        > row
            .get("max_attempts")
            .and_then(Value::as_i64)
            .ok_or(DatabaseError::StorageFailure)?
    {
        return Err(DatabaseError::Conflict);
    }
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| DatabaseError::Unavailable)?;
    let epoch = i64::try_from(now.as_secs()).map_err(|_| DatabaseError::InvalidInput)?;
    let completed = chrono::DateTime::<chrono::Utc>::from_timestamp(epoch, 0)
        .ok_or(DatabaseError::InvalidInput)?
        .to_rfc3339();
    let workflow_id = row.get("workflow_execution_id").and_then(Value::as_i64);
    let job_id = row
        .get("job_id")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let mut steps = vec![Mutation::Update {
        entity: "job_execution".to_owned(),
        filter: Predicate::All(vec![
            Predicate::Equal("id".to_owned(), Value::from(id)),
            Predicate::Equal(
                "execution_status".to_owned(),
                Value::String("queued".to_owned()),
            ),
            Predicate::Equal("attempt_count".to_owned(), Value::from(attempts - 1)),
        ]),
        values: BTreeMap::from([
            (
                "execution_status".to_owned(),
                WriteValue::Literal(Value::String("succeeded".to_owned())),
            ),
            (
                "completed_at".to_owned(),
                WriteValue::Literal(Value::String(completed)),
            ),
            (
                "attempt_count".to_owned(),
                WriteValue::Literal(Value::from(attempts)),
            ),
        ]),
        expected_rows: Some(1),
    }];
    if schema
        .entities
        .iter()
        .any(|entity| entity.name == "job_execution_attempt")
    {
        steps.push(Mutation::Insert {
            entity: "job_execution_attempt".to_owned(),
            values: BTreeMap::from([
                (
                    "job_execution_id".to_owned(),
                    WriteValue::Literal(Value::from(id)),
                ),
                (
                    "attempt_number".to_owned(),
                    WriteValue::Literal(Value::from(attempts)),
                ),
                (
                    "execution_status".to_owned(),
                    WriteValue::Literal(Value::String("succeeded".to_owned())),
                ),
                (
                    "payload_json".to_owned(),
                    WriteValue::Literal(Value::String("{}".to_owned())),
                ),
            ]),
        });
    }
    if let Some(workflow_id) = workflow_id.filter(|id| *id > 0)
        && schema
            .entities
            .iter()
            .any(|entity| entity.name == "workflow_execution_event")
    {
        steps.push(Mutation::Insert {
            entity: "workflow_execution_event".to_owned(),
            values: BTreeMap::from([
                (
                    "workflow_execution_id".to_owned(),
                    WriteValue::Literal(Value::from(workflow_id)),
                ),
                (
                    "event_kind".to_owned(),
                    WriteValue::Literal(Value::String("job_succeeded".to_owned())),
                ),
                (
                    "event_key".to_owned(),
                    WriteValue::Literal(Value::String(job_id.to_owned())),
                ),
                (
                    "payload_json".to_owned(),
                    WriteValue::Literal(Value::String("{}".to_owned())),
                ),
            ]),
        });
    }
    adapter.transact(
        config,
        schema,
        &Transaction {
            steps,
            idempotency_key: None,
        },
    )?;
    Ok(Some(id))
}

/// Reads the bounded operator job list and complete status counts.
pub fn local_job_report(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
) -> Result<(Vec<Value>, Value), DatabaseError> {
    let rows = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: [
                "id",
                "job_id",
                "handler_id",
                "queue_name",
                "execution_status",
                "attempt_count",
            ]
            .into_iter()
            .map(str::to_owned)
            .collect(),
            filter: None,
            order: vec![SortKey {
                field: "id".to_owned(),
                descending: false,
            }],
            limit: 256,
            offset: 0,
        },
    )?;
    let jobs = rows
        .into_iter()
        .map(|mut row| {
            if let Some(id) = row.remove("id") {
                row.insert("execution_id".to_owned(), id);
            }
            Value::Object(row)
        })
        .collect();
    let mut counts = serde_json::Map::new();
    for status in ["queued", "running", "succeeded", "failed"] {
        let count = adapter.aggregate(
            config,
            schema,
            &AggregateQuery {
                entity: "job_execution".to_owned(),
                filter: Some(Predicate::Equal(
                    "execution_status".to_owned(),
                    Value::String(status.to_owned()),
                )),
                aggregate: Aggregate::Count,
            },
        )?;
        counts.insert(status.to_owned(), count);
    }
    Ok((jobs, Value::Object(counts)))
}

/// Lists bounded workflow executions for the operator jobs report.
pub fn local_workflow_report(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
) -> Result<Vec<Value>, DatabaseError> {
    adapter
        .query(
            config,
            schema,
            &Query {
                entity: "workflow_execution".to_owned(),
                fields: ["id", "workflow_id", "execution_status", "replay_trace_id"]
                    .into_iter()
                    .map(str::to_owned)
                    .collect(),
                filter: None,
                order: vec![SortKey {
                    field: "id".to_owned(),
                    descending: false,
                }],
                limit: 256,
                offset: 0,
            },
        )
        .map(|rows| {
            rows.into_iter()
                .map(|mut row| {
                    if let Some(id) = row.remove("id") {
                        row.insert("execution_id".to_owned(), id);
                    }
                    Value::Object(row)
                })
                .collect()
        })
}

/// Moves bounded failed or expired-lease jobs back to the queue. The status
/// guard prevents an operator command from overwriting a concurrent worker.
pub fn requeue_local_jobs(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    retry_failed: bool,
    limit: u32,
) -> Result<usize, DatabaseError> {
    if limit == 0 || limit > 256 {
        return Err(DatabaseError::InvalidInput);
    }
    let status = if retry_failed { "failed" } else { "running" };
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| DatabaseError::Unavailable)?;
    let epoch = i64::try_from(now.as_secs()).map_err(|_| DatabaseError::InvalidInput)?;
    let timestamp = chrono::DateTime::<chrono::Utc>::from_timestamp(epoch, 0)
        .ok_or(DatabaseError::InvalidInput)?
        .to_rfc3339();
    let mut processed = 0;
    let mut last_id = 0;
    loop {
        if processed >= limit as usize {
            break;
        }
        let rows = adapter.query(
            config,
            schema,
            &Query {
                entity: "job_execution".to_owned(),
                fields: ["id", "attempt_count", "max_attempts", "lease_expires_at"]
                    .into_iter()
                    .map(str::to_owned)
                    .collect(),
                filter: Some(Predicate::All(vec![
                    Predicate::Equal(
                        "execution_status".to_owned(),
                        Value::String(status.to_owned()),
                    ),
                    Predicate::GreaterThan("id".to_owned(), Value::from(last_id)),
                    if retry_failed {
                        Predicate::GreaterThan("max_attempts".to_owned(), Value::from(0))
                    } else {
                        Predicate::LessThan(
                            "lease_expires_at".to_owned(),
                            Value::String(timestamp.clone()),
                        )
                    },
                ])),
                order: vec![SortKey {
                    field: "id".to_owned(),
                    descending: false,
                }],
                limit: 256,
                offset: 0,
            },
        )?;
        if rows.is_empty() {
            break;
        }
        let page_len = rows.len();
        for row in rows {
            let id = row
                .get("id")
                .and_then(Value::as_i64)
                .ok_or(DatabaseError::StorageFailure)?;
            last_id = id;
            let attempt_count = row
                .get("attempt_count")
                .and_then(Value::as_i64)
                .ok_or(DatabaseError::StorageFailure)?;
            let max_attempts = row
                .get("max_attempts")
                .and_then(Value::as_i64)
                .ok_or(DatabaseError::StorageFailure)?;
            if retry_failed && attempt_count >= max_attempts {
                continue;
            }
            if !retry_failed {
                let expired = row
                    .get("lease_expires_at")
                    .and_then(Value::as_str)
                    .and_then(|lease| {
                        lease.parse::<i64>().ok().or_else(|| {
                            chrono::DateTime::parse_from_rfc3339(lease)
                                .ok()
                                .map(|time| time.timestamp())
                        })
                    })
                    .is_some_and(|lease| lease < epoch);
                if !expired {
                    continue;
                }
            }
            adapter.transact(
                config,
                schema,
                &Transaction {
                    steps: vec![Mutation::Update {
                        entity: "job_execution".to_owned(),
                        filter: Predicate::All(vec![
                            Predicate::Equal("id".to_owned(), Value::from(id)),
                            Predicate::Equal(
                                "execution_status".to_owned(),
                                Value::String(status.to_owned()),
                            ),
                            Predicate::Equal(
                                "attempt_count".to_owned(),
                                Value::from(attempt_count),
                            ),
                            if retry_failed {
                                Predicate::Equal(
                                    "max_attempts".to_owned(),
                                    Value::from(max_attempts),
                                )
                            } else {
                                Predicate::Equal(
                                    "lease_expires_at".to_owned(),
                                    row.get("lease_expires_at")
                                        .cloned()
                                        .ok_or(DatabaseError::StorageFailure)?,
                                )
                            },
                        ]),
                        values: BTreeMap::from([
                            (
                                "execution_status".to_owned(),
                                WriteValue::Literal(Value::String("queued".to_owned())),
                            ),
                            ("claimed_at".to_owned(), WriteValue::Literal(Value::Null)),
                            (
                                "lease_expires_at".to_owned(),
                                WriteValue::Literal(Value::Null),
                            ),
                            (
                                "available_at".to_owned(),
                                WriteValue::Literal(Value::String(timestamp.clone())),
                            ),
                        ]),
                        expected_rows: Some(1),
                    }],
                    idempotency_key: None,
                },
            )?;
            processed += 1;
            if processed >= limit as usize {
                break;
            }
        }
        if page_len < 256 {
            break;
        }
    }
    Ok(processed)
}

/// Reads one job and its bounded attempt/event history through the adapter.
pub fn local_job_replay(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    execution_id: i64,
) -> Result<Value, DatabaseError> {
    if execution_id <= 0 {
        return Err(DatabaseError::InvalidInput);
    }
    let jobs = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution".to_owned(),
            fields: [
                "id",
                "job_id",
                "queue_name",
                "execution_status",
                "workflow_execution_id",
                "replay_trace_id",
            ]
            .into_iter()
            .map(str::to_owned)
            .collect(),
            filter: Some(Predicate::Equal("id".to_owned(), Value::from(execution_id))),
            order: vec![],
            limit: 1,
            offset: 0,
        },
    )?;
    let Some(job) = jobs.first() else {
        return Err(DatabaseError::MissingRecord);
    };
    let trace = job.get("replay_trace_id").cloned().unwrap_or(Value::Null);
    let workflow_id = job.get("workflow_execution_id").and_then(Value::as_i64);
    let attempts = adapter.query(
        config,
        schema,
        &Query {
            entity: "job_execution_attempt".to_owned(),
            fields: ["attempt_number", "execution_status", "error_message"]
                .into_iter()
                .map(str::to_owned)
                .collect(),
            filter: Some(Predicate::Equal(
                "job_execution_id".to_owned(),
                Value::from(execution_id),
            )),
            order: vec![SortKey {
                field: "attempt_number".to_owned(),
                descending: false,
            }],
            limit: 256,
            offset: 0,
        },
    )?;
    let events = if let Some(workflow_id) = workflow_id.filter(|id| *id > 0) {
        adapter.query(
            config,
            schema,
            &Query {
                entity: "workflow_execution_event".to_owned(),
                fields: ["event_kind", "event_key", "payload_json"]
                    .into_iter()
                    .map(str::to_owned)
                    .collect(),
                filter: Some(Predicate::Equal(
                    "workflow_execution_id".to_owned(),
                    Value::from(workflow_id),
                )),
                order: vec![SortKey {
                    field: "id".to_owned(),
                    descending: false,
                }],
                limit: 256,
                offset: 0,
            },
        )?
    } else {
        Vec::new()
    };
    let mut job = job.clone();
    job.remove("workflow_execution_id");
    job.remove("replay_trace_id");
    if let Some(id) = job.remove("id") {
        job.insert("execution_id".to_owned(), id);
    }
    Ok(serde_json::json!({
        "replay_trace_id": trace,
        "jobs": [job],
        "attempts": attempts,
        "events": events,
    }))
}