Genarrative/server-rs/crates/spacetime-module/src/external_generation.rs

use crate::*;

const EXTERNAL_GENERATION_STATUS_PENDING: &str = "pending";
const EXTERNAL_GENERATION_STATUS_RUNNING: &str = "running";
const EXTERNAL_GENERATION_STATUS_COMPLETED: &str = "completed";
const EXTERNAL_GENERATION_STATUS_FAILED: &str = "failed";
const EXTERNAL_GENERATION_STATUS_CANCELLED: &str = "cancelled";

#[spacetimedb::table(
    accessor = external_generation_job,
    index(
        accessor = by_external_generation_job_status_available,
        btree(columns = [status, available_at])
    ),
    index(
        accessor = by_external_generation_job_worker_id,
        btree(columns = [worker_id])
    ),
    index(
        accessor = by_external_generation_job_source,
        btree(columns = [source_module, source_entity_id])
    ),
    index(
        accessor = by_external_generation_job_owner_user_id,
        btree(columns = [owner_user_id])
    )
)]
#[derive(Clone)]
pub struct ExternalGenerationJob {
    #[primary_key]
    pub(crate) job_id: String,
    #[unique]
    pub(crate) dedupe_key: String,
    pub(crate) job_kind: String,
    pub(crate) owner_user_id: String,
    pub(crate) source_module: String,
    pub(crate) source_entity_id: String,
    pub(crate) request_label: String,
    pub(crate) request_payload_json: String,
    pub(crate) status: String,
    pub(crate) attempt: u32,
    pub(crate) max_attempts: u32,
    pub(crate) last_error_message: Option<String>,
    pub(crate) worker_id: Option<String>,
    pub(crate) lease_expires_at: Option<Timestamp>,
    pub(crate) available_at: Timestamp,
    pub(crate) result_payload_json: Option<String>,
    pub(crate) created_at: Timestamp,
    pub(crate) started_at: Option<Timestamp>,
    pub(crate) completed_at: Option<Timestamp>,
    pub(crate) updated_at: Timestamp,
    #[default(None::<String>)]
    pub(crate) lease_token: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobEnqueueInput {
    pub job_id: String,
    pub dedupe_key: String,
    pub job_kind: String,
    pub owner_user_id: String,
    pub source_module: String,
    pub source_entity_id: String,
    pub request_label: String,
    pub request_payload_json: String,
    pub max_attempts: u32,
    pub available_at_micros: i64,
    pub created_at_micros: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobClaimInput {
    pub worker_id: String,
    pub limit: u32,
    pub lease_expires_at_micros: i64,
    pub claimed_at_micros: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobRenewLeaseInput {
    pub job_id: String,
    pub worker_id: String,
    pub lease_token: String,
    pub lease_expires_at_micros: i64,
    pub renewed_at_micros: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobCompleteInput {
    pub job_id: String,
    pub worker_id: String,
    pub lease_token: String,
    pub result_payload_json: Option<String>,
    pub completed_at_micros: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobFailInput {
    pub job_id: String,
    pub worker_id: String,
    pub lease_token: String,
    pub error_message: String,
    pub retry_after_micros: i64,
    pub failed_at_micros: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobSnapshot {
    pub job_id: String,
    pub dedupe_key: String,
    pub job_kind: String,
    pub owner_user_id: String,
    pub source_module: String,
    pub source_entity_id: String,
    pub request_label: String,
    pub request_payload_json: String,
    pub status: String,
    pub attempt: u32,
    pub max_attempts: u32,
    pub last_error_message: Option<String>,
    pub worker_id: Option<String>,
    pub lease_expires_at_micros: Option<i64>,
    pub available_at_micros: i64,
    pub result_payload_json: Option<String>,
    pub created_at_micros: i64,
    pub started_at_micros: Option<i64>,
    pub completed_at_micros: Option<i64>,
    pub updated_at_micros: i64,
    pub lease_token: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobProcedureResult {
    pub ok: bool,
    pub job: Option<ExternalGenerationJobSnapshot>,
    pub jobs: Vec<ExternalGenerationJobSnapshot>,
    pub error_message: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationQueueStatsSnapshot {
    pub pending_count: u32,
    pub delayed_pending_count: u32,
    pub claimable_pending_count: u32,
    pub running_active_count: u32,
    pub expired_running_count: u32,
    // 中文注释：保留字段兼容已生成 bindings；controller 只按非终态队列压力扩缩容，不每轮扫描历史终态任务。
    pub terminal_count: u32,
    pub claimable_count: u32,
    pub oldest_claimable_age_micros: Option<i64>,
    pub now_micros: i64,
}

#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationQueueStatsProcedureResult {
    pub ok: bool,
    pub stats: Option<ExternalGenerationQueueStatsSnapshot>,
    pub error_message: Option<String>,
}

#[spacetimedb::procedure]
pub fn enqueue_external_generation_job_and_return(
    ctx: &mut ProcedureContext,
    input: ExternalGenerationJobEnqueueInput,
) -> ExternalGenerationJobProcedureResult {
    match ctx.try_with_tx(|tx| enqueue_external_generation_job_tx(tx, input.clone())) {
        Ok(job) => single_external_generation_job_result(job),
        Err(message) => failed_external_generation_job_result(message),
    }
}

#[spacetimedb::procedure]
pub fn claim_external_generation_jobs_and_return(
    ctx: &mut ProcedureContext,
    input: ExternalGenerationJobClaimInput,
) -> ExternalGenerationJobProcedureResult {
    match ctx.try_with_tx(|tx| claim_external_generation_jobs_tx(tx, input.clone())) {
        Ok(jobs) => ExternalGenerationJobProcedureResult {
            ok: true,
            job: None,
            jobs,
            error_message: None,
        },
        Err(message) => failed_external_generation_job_result(message),
    }
}

#[spacetimedb::procedure]
pub fn complete_external_generation_job_and_return(
    ctx: &mut ProcedureContext,
    input: ExternalGenerationJobCompleteInput,
) -> ExternalGenerationJobProcedureResult {
    match ctx.try_with_tx(|tx| complete_external_generation_job_tx(tx, input.clone())) {
        Ok(job) => single_external_generation_job_result(job),
        Err(message) => failed_external_generation_job_result(message),
    }
}

#[spacetimedb::procedure]
pub fn renew_external_generation_job_lease_and_return(
    ctx: &mut ProcedureContext,
    input: ExternalGenerationJobRenewLeaseInput,
) -> ExternalGenerationJobProcedureResult {
    match ctx.try_with_tx(|tx| renew_external_generation_job_lease_tx(tx, input.clone())) {
        Ok(job) => single_external_generation_job_result(job),
        Err(message) => failed_external_generation_job_result(message),
    }
}

#[spacetimedb::procedure]
pub fn fail_external_generation_job_and_return(
    ctx: &mut ProcedureContext,
    input: ExternalGenerationJobFailInput,
) -> ExternalGenerationJobProcedureResult {
    match ctx.try_with_tx(|tx| fail_external_generation_job_tx(tx, input.clone())) {
        Ok(job) => single_external_generation_job_result(job),
        Err(message) => failed_external_generation_job_result(message),
    }
}

#[spacetimedb::procedure]
pub fn get_external_generation_queue_stats_and_return(
    ctx: &mut ProcedureContext,
) -> ExternalGenerationQueueStatsProcedureResult {
    match ctx.try_with_tx(|tx| get_external_generation_queue_stats_tx(tx)) {
        Ok(stats) => ExternalGenerationQueueStatsProcedureResult {
            ok: true,
            stats: Some(stats),
            error_message: None,
        },
        Err(message) => ExternalGenerationQueueStatsProcedureResult {
            ok: false,
            stats: None,
            error_message: Some(message),
        },
    }
}

fn enqueue_external_generation_job_tx(
    ctx: &ReducerContext,
    input: ExternalGenerationJobEnqueueInput,
) -> Result<ExternalGenerationJobSnapshot, String> {
    validate_required("external_generation_job.job_id", &input.job_id)?;
    validate_required("external_generation_job.dedupe_key", &input.dedupe_key)?;
    validate_required("external_generation_job.job_kind", &input.job_kind)?;
    validate_required(
        "external_generation_job.owner_user_id",
        &input.owner_user_id,
    )?;
    validate_required(
        "external_generation_job.source_module",
        &input.source_module,
    )?;
    validate_required(
        "external_generation_job.source_entity_id",
        &input.source_entity_id,
    )?;
    validate_required(
        "external_generation_job.request_label",
        &input.request_label,
    )?;
    validate_required(
        "external_generation_job.request_payload_json",
        &input.request_payload_json,
    )?;

    if let Some(row) = ctx
        .db
        .external_generation_job()
        .dedupe_key()
        .find(&input.dedupe_key)
    {
        return Ok(map_external_generation_job_row(row));
    }
    if ctx
        .db
        .external_generation_job()
        .job_id()
        .find(&input.job_id)
        .is_some()
    {
        return Err("external_generation_job.job_id 已存在".to_string());
    }

    let now = Timestamp::from_micros_since_unix_epoch(input.created_at_micros);
    let available_at = Timestamp::from_micros_since_unix_epoch(input.available_at_micros);
    let row = ExternalGenerationJob {
        job_id: input.job_id.trim().to_string(),
        dedupe_key: input.dedupe_key.trim().to_string(),
        job_kind: input.job_kind.trim().to_string(),
        owner_user_id: input.owner_user_id.trim().to_string(),
        source_module: input.source_module.trim().to_string(),
        source_entity_id: input.source_entity_id.trim().to_string(),
        request_label: input.request_label.trim().to_string(),
        request_payload_json: input.request_payload_json.trim().to_string(),
        status: EXTERNAL_GENERATION_STATUS_PENDING.to_string(),
        attempt: 0,
        max_attempts: input.max_attempts.max(1),
        last_error_message: None,
        worker_id: None,
        lease_expires_at: None,
        available_at,
        result_payload_json: None,
        created_at: now,
        started_at: None,
        completed_at: None,
        updated_at: now,
        lease_token: None,
    };
    ctx.db.external_generation_job().insert(row.clone());
    Ok(map_external_generation_job_row(row))
}

fn claim_external_generation_jobs_tx(
    ctx: &ReducerContext,
    input: ExternalGenerationJobClaimInput,
) -> Result<Vec<ExternalGenerationJobSnapshot>, String> {
    validate_required("external_generation_job.worker_id", &input.worker_id)?;
    if input.limit == 0 {
        return Ok(Vec::new());
    }

    let claim_time = ctx.timestamp;
    let lease_duration_micros = duration_between_micros(
        input.lease_expires_at_micros,
        input.claimed_at_micros,
        "external_generation_job.lease_duration",
    )?;
    let lease_expires_at = timestamp_after_micros(claim_time, lease_duration_micros);
    let worker_id = input.worker_id.trim().to_string();
    let limit = input.limit.min(64) as usize;
    let mut candidates = Vec::new();

    candidates.extend(
        ctx.db
            .external_generation_job()
            .by_external_generation_job_status_available()
            .filter(&EXTERNAL_GENERATION_STATUS_PENDING.to_string())
            .filter(|row| is_external_generation_job_claimable(row, claim_time)),
    );
    candidates.extend(
        ctx.db
            .external_generation_job()
            .by_external_generation_job_status_available()
            .filter(&EXTERNAL_GENERATION_STATUS_RUNNING.to_string())
            .filter(|row| is_external_generation_job_claimable(row, claim_time)),
    );

    candidates.sort_by(|left, right| {
        left.available_at
            .to_micros_since_unix_epoch()
            .cmp(&right.available_at.to_micros_since_unix_epoch())
            .then_with(|| {
                left.created_at
                    .to_micros_since_unix_epoch()
                    .cmp(&right.created_at.to_micros_since_unix_epoch())
            })
            .then_with(|| left.job_id.cmp(&right.job_id))
    });

    let mut claimed = Vec::new();
    for mut row in candidates.into_iter().take(limit) {
        let next_attempt = row.attempt.saturating_add(1);
        let lease_token = build_external_generation_lease_token(
            &row.job_id,
            &worker_id,
            next_attempt,
            claim_time,
        );
        row.status = EXTERNAL_GENERATION_STATUS_RUNNING.to_string();
        row.worker_id = Some(worker_id.clone());
        row.lease_expires_at = Some(lease_expires_at);
        row.lease_token = Some(lease_token);
        row.attempt = next_attempt;
        if row.started_at.is_none() {
            row.started_at = Some(claim_time);
        }
        row.updated_at = claim_time;
        persist_external_generation_job_row(ctx, row.clone());
        claimed.push(map_external_generation_job_row(row));
    }

    Ok(claimed)
}

fn complete_external_generation_job_tx(
    ctx: &ReducerContext,
    input: ExternalGenerationJobCompleteInput,
) -> Result<ExternalGenerationJobSnapshot, String> {
    let mut row = get_worker_owned_external_generation_job(
        ctx,
        &input.job_id,
        &input.worker_id,
        &input.lease_token,
    )?;
    let completed_at = ctx.timestamp;
    row.status = EXTERNAL_GENERATION_STATUS_COMPLETED.to_string();
    row.result_payload_json = input
        .result_payload_json
        .and_then(|value| normalize_optional_text(value.as_str()));
    row.lease_expires_at = None;
    row.completed_at = Some(completed_at);
    row.updated_at = completed_at;
    persist_external_generation_job_row(ctx, row.clone());
    Ok(map_external_generation_job_row(row))
}

fn renew_external_generation_job_lease_tx(
    ctx: &ReducerContext,
    input: ExternalGenerationJobRenewLeaseInput,
) -> Result<ExternalGenerationJobSnapshot, String> {
    let mut row = get_worker_owned_external_generation_job(
        ctx,
        &input.job_id,
        &input.worker_id,
        &input.lease_token,
    )?;
    let renewed_at = ctx.timestamp;
    let lease_duration_micros = duration_between_micros(
        input.lease_expires_at_micros,
        input.renewed_at_micros,
        "external_generation_job.lease_duration",
    )?;
    row.lease_expires_at = Some(timestamp_after_micros(renewed_at, lease_duration_micros));
    row.updated_at = renewed_at;
    persist_external_generation_job_row(ctx, row.clone());
    Ok(map_external_generation_job_row(row))
}

fn fail_external_generation_job_tx(
    ctx: &ReducerContext,
    input: ExternalGenerationJobFailInput,
) -> Result<ExternalGenerationJobSnapshot, String> {
    let error_message = input.error_message.trim();
    if error_message.is_empty() {
        return Err("external_generation_job.error_message 不能为空".to_string());
    }

    let mut row = get_worker_owned_external_generation_job(
        ctx,
        &input.job_id,
        &input.worker_id,
        &input.lease_token,
    )?;
    let failed_at = ctx.timestamp;
    let retry_delay_micros = duration_between_micros(
        input.retry_after_micros,
        input.failed_at_micros,
        "external_generation_job.retry_delay",
    )?;
    row.last_error_message = Some(error_message.to_string());
    row.lease_expires_at = None;
    row.worker_id = None;
    row.lease_token = None;
    row.updated_at = failed_at;

    if row.attempt < row.max_attempts {
        row.status = EXTERNAL_GENERATION_STATUS_PENDING.to_string();
        row.available_at = timestamp_after_micros(failed_at, retry_delay_micros);
    } else {
        row.status = EXTERNAL_GENERATION_STATUS_FAILED.to_string();
        row.completed_at = Some(failed_at);
    }

    persist_external_generation_job_row(ctx, row.clone());
    Ok(map_external_generation_job_row(row))
}

fn get_external_generation_queue_stats_tx(
    ctx: &ReducerContext,
) -> Result<ExternalGenerationQueueStatsSnapshot, String> {
    let now = ctx.timestamp;
    let now_micros = now.to_micros_since_unix_epoch();
    let mut stats = ExternalGenerationQueueStatsSnapshot {
        pending_count: 0,
        delayed_pending_count: 0,
        claimable_pending_count: 0,
        running_active_count: 0,
        expired_running_count: 0,
        terminal_count: 0,
        claimable_count: 0,
        oldest_claimable_age_micros: None,
        now_micros,
    };

    for row in ctx
        .db
        .external_generation_job()
        .by_external_generation_job_status_available()
        .filter(&EXTERNAL_GENERATION_STATUS_PENDING.to_string())
    {
        stats.pending_count = stats.pending_count.saturating_add(1);
        if is_external_generation_job_claimable(&row, now) {
            stats.claimable_pending_count = stats.claimable_pending_count.saturating_add(1);
            record_external_generation_claimable_age(&mut stats, &row, now_micros);
        } else {
            stats.delayed_pending_count = stats.delayed_pending_count.saturating_add(1);
        }
    }

    for row in ctx
        .db
        .external_generation_job()
        .by_external_generation_job_status_available()
        .filter(&EXTERNAL_GENERATION_STATUS_RUNNING.to_string())
    {
        if is_external_generation_job_claimable(&row, now) {
            stats.expired_running_count = stats.expired_running_count.saturating_add(1);
            record_external_generation_claimable_age(&mut stats, &row, now_micros);
        } else {
            stats.running_active_count = stats.running_active_count.saturating_add(1);
        }
    }

    stats.claimable_count = stats
        .claimable_pending_count
        .saturating_add(stats.expired_running_count);
    Ok(stats)
}

pub(crate) fn validate_external_generation_job_lease_for_tx(
    ctx: &ReducerContext,
    job_id: &str,
    worker_id: &str,
    lease_token: &str,
    expected_job_kinds: &[&str],
    expected_owner_user_id: &str,
    expected_source_module: &str,
    expected_source_entity_ids: &[String],
) -> Result<(), String> {
    let row = get_worker_owned_external_generation_job(ctx, job_id, worker_id, lease_token)?;
    if !expected_job_kinds.is_empty()
        && !expected_job_kinds
            .iter()
            .any(|expected| row.job_kind.trim() == expected.trim())
    {
        return Err("external_generation_job job_kind 与业务写回不匹配".to_string());
    }
    if row.owner_user_id.trim() != expected_owner_user_id.trim() {
        return Err("external_generation_job owner_user_id 与业务写回不匹配".to_string());
    }
    if row.source_module.trim() != expected_source_module.trim() {
        return Err("external_generation_job source_module 与业务写回不匹配".to_string());
    }
    if !expected_source_entity_ids
        .iter()
        .any(|expected| row.source_entity_id.trim() == expected.trim())
    {
        return Err("external_generation_job source_entity_id 与业务写回不匹配".to_string());
    }
    Ok(())
}

fn get_worker_owned_external_generation_job(
    ctx: &ReducerContext,
    job_id: &str,
    worker_id: &str,
    lease_token: &str,
) -> Result<ExternalGenerationJob, String> {
    validate_required("external_generation_job.job_id", job_id)?;
    validate_required("external_generation_job.worker_id", worker_id)?;
    validate_required("external_generation_job.lease_token", lease_token)?;
    let row = ctx
        .db
        .external_generation_job()
        .job_id()
        .find(&job_id.trim().to_string())
        .ok_or_else(|| "external_generation_job 不存在".to_string())?;
    if row.status != EXTERNAL_GENERATION_STATUS_RUNNING {
        return Err("external_generation_job 当前不是 running 状态".to_string());
    }
    if !is_external_generation_job_owned_by_worker(&row, worker_id) {
        return Err("external_generation_job worker lease 不匹配".to_string());
    }
    if !is_external_generation_job_owned_by_lease_token(&row, lease_token) {
        return Err("external_generation_job lease token 不匹配".to_string());
    }
    if !is_external_generation_job_lease_active(&row, ctx.timestamp) {
        return Err("external_generation_job lease 已过期".to_string());
    }
    Ok(row)
}

fn is_external_generation_job_owned_by_worker(
    row: &ExternalGenerationJob,
    worker_id: &str,
) -> bool {
    row.worker_id.as_deref() == Some(worker_id.trim())
}

fn is_external_generation_job_owned_by_lease_token(
    row: &ExternalGenerationJob,
    lease_token: &str,
) -> bool {
    row.lease_token.as_deref() == Some(lease_token.trim())
}

fn is_external_generation_job_lease_active(row: &ExternalGenerationJob, now: Timestamp) -> bool {
    row.lease_expires_at
        .map(|lease_expires_at| lease_expires_at > now)
        .unwrap_or(false)
}

fn is_external_generation_job_claimable(row: &ExternalGenerationJob, now: Timestamp) -> bool {
    match row.status.as_str() {
        EXTERNAL_GENERATION_STATUS_PENDING => row.available_at <= now,
        EXTERNAL_GENERATION_STATUS_RUNNING => row
            .lease_expires_at
            .map(|lease_expires_at| lease_expires_at <= now)
            .unwrap_or(true),
        EXTERNAL_GENERATION_STATUS_COMPLETED
        | EXTERNAL_GENERATION_STATUS_FAILED
        | EXTERNAL_GENERATION_STATUS_CANCELLED => false,
        _ => false,
    }
}

fn record_external_generation_claimable_age(
    stats: &mut ExternalGenerationQueueStatsSnapshot,
    row: &ExternalGenerationJob,
    now_micros: i64,
) {
    let age = now_micros
        .saturating_sub(row.available_at.to_micros_since_unix_epoch())
        .max(0);
    stats.oldest_claimable_age_micros = Some(
        stats
            .oldest_claimable_age_micros
            .map(|current| current.max(age))
            .unwrap_or(age),
    );
}

fn persist_external_generation_job_row(ctx: &ReducerContext, row: ExternalGenerationJob) {
    ctx.db
        .external_generation_job()
        .job_id()
        .delete(&row.job_id);
    ctx.db.external_generation_job().insert(row);
}

fn map_external_generation_job_row(row: ExternalGenerationJob) -> ExternalGenerationJobSnapshot {
    ExternalGenerationJobSnapshot {
        job_id: row.job_id,
        dedupe_key: row.dedupe_key,
        job_kind: row.job_kind,
        owner_user_id: row.owner_user_id,
        source_module: row.source_module,
        source_entity_id: row.source_entity_id,
        request_label: row.request_label,
        request_payload_json: row.request_payload_json,
        status: row.status,
        attempt: row.attempt,
        max_attempts: row.max_attempts,
        last_error_message: row.last_error_message,
        worker_id: row.worker_id,
        lease_expires_at_micros: row
            .lease_expires_at
            .map(|value| value.to_micros_since_unix_epoch()),
        available_at_micros: row.available_at.to_micros_since_unix_epoch(),
        result_payload_json: row.result_payload_json,
        created_at_micros: row.created_at.to_micros_since_unix_epoch(),
        started_at_micros: row
            .started_at
            .map(|value| value.to_micros_since_unix_epoch()),
        completed_at_micros: row
            .completed_at
            .map(|value| value.to_micros_since_unix_epoch()),
        updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
        lease_token: row.lease_token,
    }
}

fn single_external_generation_job_result(
    job: ExternalGenerationJobSnapshot,
) -> ExternalGenerationJobProcedureResult {
    ExternalGenerationJobProcedureResult {
        ok: true,
        job: Some(job),
        jobs: Vec::new(),
        error_message: None,
    }
}

fn failed_external_generation_job_result(message: String) -> ExternalGenerationJobProcedureResult {
    ExternalGenerationJobProcedureResult {
        ok: false,
        job: None,
        jobs: Vec::new(),
        error_message: Some(message),
    }
}

fn validate_required(field: &str, value: &str) -> Result<(), String> {
    if value.trim().is_empty() {
        return Err(format!("{field} 不能为空"));
    }
    Ok(())
}

fn duration_between_micros(later: i64, earlier: i64, field: &str) -> Result<i64, String> {
    let duration = later.saturating_sub(earlier);
    if duration <= 0 {
        return Err(format!("{field} 必须大于 0"));
    }
    Ok(duration)
}

fn timestamp_after_micros(timestamp: Timestamp, duration_micros: i64) -> Timestamp {
    Timestamp::from_micros_since_unix_epoch(
        timestamp
            .to_micros_since_unix_epoch()
            .saturating_add(duration_micros.max(0)),
    )
}

fn build_external_generation_lease_token(
    job_id: &str,
    worker_id: &str,
    attempt: u32,
    claimed_at: Timestamp,
) -> String {
    format!(
        "{}:{}:{}:{}",
        job_id.trim(),
        worker_id.trim(),
        attempt,
        claimed_at.to_micros_since_unix_epoch()
    )
}

fn normalize_optional_text(value: &str) -> Option<String> {
    let normalized = value.trim();
    if normalized.is_empty() {
        None
    } else {
        Some(normalized.to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn external_generation_job_result_failure_is_structured() {
        let result = failed_external_generation_job_result("失败".to_string());
        assert!(!result.ok);
        assert_eq!(result.error_message.as_deref(), Some("失败"));
        assert!(result.jobs.is_empty());
    }

    #[test]
    fn pending_job_is_claimable_only_after_available_time() {
        let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_PENDING);
        row.available_at = micros(1_000);

        assert!(!is_external_generation_job_claimable(&row, micros(999)));
        assert!(is_external_generation_job_claimable(&row, micros(1_000)));
    }

    #[test]
    fn running_job_is_claimable_only_after_lease_expires() {
        let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
        row.lease_expires_at = Some(micros(2_000));

        assert!(!is_external_generation_job_claimable(&row, micros(1_999)));
        assert!(is_external_generation_job_claimable(&row, micros(2_000)));
    }

    #[test]
    fn terminal_job_is_not_claimable() {
        for status in [
            EXTERNAL_GENERATION_STATUS_COMPLETED,
            EXTERNAL_GENERATION_STATUS_FAILED,
            EXTERNAL_GENERATION_STATUS_CANCELLED,
        ] {
            let row = external_generation_job_fixture(status);

            assert!(!is_external_generation_job_claimable(&row, micros(10_000)));
        }
    }

    #[test]
    fn worker_ownership_requires_matching_trimmed_worker_id() {
        let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
        row.worker_id = Some("worker-a".to_string());

        assert!(is_external_generation_job_owned_by_worker(
            &row,
            " worker-a "
        ));
        assert!(!is_external_generation_job_owned_by_worker(
            &row, "worker-b"
        ));
    }

    #[test]
    fn worker_ownership_requires_matching_trimmed_lease_token() {
        let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
        row.lease_token = Some("job-1:worker-a:1:1000".to_string());

        assert!(is_external_generation_job_owned_by_lease_token(
            &row,
            " job-1:worker-a:1:1000 "
        ));
        assert!(!is_external_generation_job_owned_by_lease_token(
            &row,
            "job-1:worker-a:2:2000"
        ));
    }

    #[test]
    fn worker_lease_is_active_only_before_expiry() {
        let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
        row.lease_expires_at = Some(micros(2_000));

        assert!(is_external_generation_job_lease_active(&row, micros(1_999)));
        assert!(!is_external_generation_job_lease_active(
            &row,
            micros(2_000)
        ));
    }

    #[test]
    fn lease_token_changes_with_claim_attempt() {
        let first =
            build_external_generation_lease_token("extgen-test", "worker-a", 1, micros(1_000));
        let second =
            build_external_generation_lease_token("extgen-test", "worker-a", 2, micros(2_000));

        assert_ne!(first, second);
    }

    #[test]
    fn claimable_age_keeps_oldest_available_job() {
        let mut stats = ExternalGenerationQueueStatsSnapshot {
            pending_count: 0,
            delayed_pending_count: 0,
            claimable_pending_count: 0,
            running_active_count: 0,
            expired_running_count: 0,
            terminal_count: 0,
            claimable_count: 0,
            oldest_claimable_age_micros: None,
            now_micros: 10_000,
        };
        let mut old_job = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_PENDING);
        old_job.available_at = micros(1_000);
        let mut newer_job = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
        newer_job.available_at = micros(8_000);

        record_external_generation_claimable_age(&mut stats, &newer_job, 10_000);
        record_external_generation_claimable_age(&mut stats, &old_job, 10_000);

        assert_eq!(stats.oldest_claimable_age_micros, Some(9_000));
    }

    #[test]
    fn positive_duration_between_client_times_is_preserved() {
        assert_eq!(
            duration_between_micros(3_500, 1_000, "external_generation_job.lease_duration"),
            Ok(2_500),
        );
        assert!(duration_between_micros(1_000, 1_000, "duration").is_err());
    }

    fn external_generation_job_fixture(status: &str) -> ExternalGenerationJob {
        ExternalGenerationJob {
            job_id: "extgen-test".to_string(),
            dedupe_key: "puzzle:compile:test".to_string(),
            job_kind: "puzzle_compile_draft".to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: "puzzle".to_string(),
            source_entity_id: "session-1".to_string(),
            request_label: "拼图首关草稿生成".to_string(),
            request_payload_json: r#"{"sessionId":"session-1"}"#.to_string(),
            status: status.to_string(),
            attempt: 0,
            max_attempts: 1,
            last_error_message: None,
            worker_id: None,
            lease_expires_at: None,
            available_at: micros(0),
            result_payload_json: None,
            created_at: micros(0),
            started_at: None,
            completed_at: None,
            updated_at: micros(0),
            lease_token: None,
        }
    }

    fn micros(value: i64) -> Timestamp {
        Timestamp::from_micros_since_unix_epoch(value)
    }
}