use crate::extraction_input_digest;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use utoipa::ToSchema;
pub const EXTRACTION_CONSOLE_PROJECTION_PROTOCOL: &str = "lenso.extraction-console.v1";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleArtifacts {
pub readiness: Option<Value>,
pub plan: Option<Value>,
#[serde(default)]
pub phase_artifacts: Vec<Value>,
pub authority: Option<Value>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionConsoleState {
Planned,
Preparing,
Blocked,
Quiesced,
Provisional,
RolledBack,
Committed,
PostCommitRollbackBlocked,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleAuthority {
pub kind: String,
pub owner_id: String,
pub revision: String,
}
impl Default for ExtractionConsoleAuthority {
fn default() -> Self {
Self {
kind: "unknown".into(),
owner_id: "unknown".into(),
revision: "unknown".into(),
}
}
}
#[derive(
Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleBlocker {
pub code: String,
pub subject: String,
pub detail: String,
#[serde(default)]
pub next_actions: Vec<String>,
pub artifact_id: String,
}
#[derive(
Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleEvidence {
pub kind: String,
pub subject: String,
pub digest: String,
pub detail: String,
pub artifact_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleTimelineEntry {
pub phase_id: String,
pub kind: String,
pub state: String,
pub artifact_id: String,
#[serde(default)]
pub blockers: Vec<ExtractionConsoleBlocker>,
#[serde(default)]
pub evidence: Vec<ExtractionConsoleEvidence>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleApprovalBoundary {
pub boundary_id: String,
pub phase_id: String,
pub action: String,
pub reason: String,
#[serde(default)]
pub required_pins: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionConsoleProjection {
pub protocol: String,
pub projection_digest: String,
pub state: ExtractionConsoleState,
pub plan_id: Option<String>,
pub plan_digest: Option<String>,
pub readiness_summary: String,
pub current_authority: ExtractionConsoleAuthority,
#[serde(default)]
pub timeline: Vec<ExtractionConsoleTimelineEntry>,
#[serde(default)]
pub blockers: Vec<ExtractionConsoleBlocker>,
#[serde(default)]
pub evidence: Vec<ExtractionConsoleEvidence>,
#[serde(default)]
pub approval_boundaries: Vec<ExtractionConsoleApprovalBoundary>,
pub read_only: bool,
#[serde(default)]
pub apply_actions: Vec<String>,
pub protected_workflow: String,
}
#[must_use]
pub fn project_extraction_console(
mut artifacts: ExtractionConsoleArtifacts,
) -> ExtractionConsoleProjection {
let plan_id = artifacts.plan.as_ref().and_then(|v| text(v, "planId"));
let plan_digest = artifacts.plan.as_ref().and_then(|v| text(v, "planDigest"));
let readiness_summary = match artifacts.readiness.as_ref() {
None => "Readiness evidence has not been recorded.".to_owned(),
Some(v) if v.get("ready").and_then(Value::as_bool) == Some(true) => {
"Extraction readiness passed with no blocking findings.".to_owned()
}
Some(v) => format!(
"Extraction readiness is blocked by {} finding(s).",
array(v, "findings").len()
),
};
let current_authority = artifacts
.authority
.as_ref()
.map(|v| ExtractionConsoleAuthority {
kind: text(v, "kind").unwrap_or_else(|| "unknown".into()),
owner_id: text(v, "ownerId").unwrap_or_else(|| "unknown".into()),
revision: text(v, "revision").unwrap_or_else(|| "unknown".into()),
})
.unwrap_or_default();
let mut timeline = planned_timeline(artifacts.plan.as_ref());
let mut blockers = blockers_from(artifacts.readiness.as_ref(), "readiness", "findings");
let mut evidence = Vec::new();
for artifact in &artifacts.phase_artifacts {
let id = artifact_id(artifact);
let mut phase_blockers = blockers_from(Some(artifact), &id, "issues");
phase_blockers.extend(blockers_from(Some(artifact), &id, "errors"));
let phase_evidence = evidence_from(artifact, &id);
blockers.extend(phase_blockers.iter().cloned());
evidence.extend(phase_evidence.iter().cloned());
timeline.push(ExtractionConsoleTimelineEntry {
phase_id: phase_for(protocol(artifact)).into(),
kind: protocol(artifact).into(),
state: status(artifact).into(),
artifact_id: id,
blockers: phase_blockers,
evidence: phase_evidence,
});
}
timeline.sort_by_key(|entry| (phase_order(&entry.phase_id), entry.artifact_id.clone()));
blockers.sort();
blockers.dedup();
evidence.sort();
evidence.dedup();
let state = state(&artifacts, !blockers.is_empty());
artifacts.phase_artifacts.sort_by_key(artifact_id);
let mut projection = ExtractionConsoleProjection {
protocol: EXTRACTION_CONSOLE_PROJECTION_PROTOCOL.into(),
projection_digest: String::new(),
state,
plan_id,
plan_digest,
readiness_summary,
current_authority,
timeline,
blockers,
evidence,
approval_boundaries: approvals(artifacts.plan.as_ref()),
read_only: true,
apply_actions: vec![],
protected_workflow: "lenso service extract".into(),
};
projection.projection_digest = digest(&projection);
projection
}
pub async fn record_extraction_artifact(
pool: &sqlx::PgPool,
plan_id: &str,
artifact: &Value,
) -> Result<(), sqlx::Error> {
let persisted = sqlx::query(
r#"
insert into platform.extraction_artifacts
(plan_id, artifact_id, protocol, artifact_digest, artifact_json)
values ($1, $2, $3, $4, $5)
on conflict (plan_id, artifact_id, artifact_digest) do nothing
"#,
)
.bind(plan_id)
.bind(artifact_id(artifact))
.bind(protocol(artifact))
.bind(digest(artifact))
.bind(artifact)
.execute(pool)
.await?;
let _ = persisted;
Ok(())
}
pub async fn load_extraction_console_projection(
pool: &sqlx::PgPool,
requested_plan_id: Option<&str>,
) -> Result<ExtractionConsoleProjection, sqlx::Error> {
let exists = sqlx::query_scalar::<_, Option<String>>(
"select to_regclass('platform.extraction_artifacts')::text",
)
.fetch_one(pool)
.await?
.is_some();
if !exists {
return Ok(empty_projection());
}
let plan_id = match requested_plan_id {
Some(plan_id) => Some(plan_id.to_owned()),
None => sqlx::query_scalar::<_, String>(
"select plan_id from platform.extraction_artifacts order by recorded_at desc, plan_id desc limit 1",
)
.fetch_optional(pool)
.await?,
};
let Some(plan_id) = plan_id else {
return Ok(empty_projection());
};
let rows = sqlx::query_as::<_, (String, Value)>(
"select protocol, artifact_json from platform.extraction_artifacts where plan_id = $1 order by recorded_at, artifact_id",
)
.bind(plan_id)
.fetch_all(pool)
.await?;
let mut artifacts = ExtractionConsoleArtifacts {
readiness: None,
plan: None,
phase_artifacts: Vec::new(),
authority: None,
};
for (protocol, artifact) in rows {
match protocol.as_str() {
"lenso.extraction-readiness-report.v1" => artifacts.readiness = Some(artifact),
"lenso.extraction-plan.v1" => artifacts.plan = Some(artifact),
"lenso.extraction-authority.v1" => artifacts.authority = Some(artifact),
_ => artifacts.phase_artifacts.push(artifact),
}
}
Ok(project_extraction_console(artifacts))
}
pub async fn load_extraction_artifact(
pool: &sqlx::PgPool,
plan_id: &str,
artifact_id: &str,
) -> Result<Option<Value>, sqlx::Error> {
sqlx::query_scalar(
"select artifact_json from platform.extraction_artifacts where plan_id = $1 and artifact_id = $2 order by recorded_at desc limit 1",
)
.bind(plan_id)
.bind(artifact_id)
.fetch_optional(pool)
.await
}
fn empty_projection() -> ExtractionConsoleProjection {
project_extraction_console(ExtractionConsoleArtifacts {
readiness: None,
plan: None,
phase_artifacts: Vec::new(),
authority: None,
})
}
fn state(a: &ExtractionConsoleArtifacts, blocked: bool) -> ExtractionConsoleState {
let has = |p: &str, s: &str| {
a.phase_artifacts
.iter()
.any(|v| protocol(v) == p && status(v) == s)
};
if a.phase_artifacts.iter().rev().any(|v| {
protocol(v) == "lenso.extraction-authority-commit.v1"
&& v.get("fastRollbackBlocked").and_then(Value::as_bool) == Some(true)
}) {
ExtractionConsoleState::PostCommitRollbackBlocked
} else if has("lenso.extraction-authority-commit.v1", "committed") {
ExtractionConsoleState::Committed
} else if let Some(cutover) = a
.phase_artifacts
.iter()
.rev()
.find(|v| protocol(v) == "lenso.extraction-provisional-cutover.v1")
{
if status(cutover) == "rolled_back" {
ExtractionConsoleState::RolledBack
} else {
ExtractionConsoleState::Provisional
}
} else if has("lenso.extraction-quiescence.v1", "quiesced") {
ExtractionConsoleState::Quiesced
} else if blocked || a.phase_artifacts.iter().any(|v| status(v) == "blocked") {
ExtractionConsoleState::Blocked
} else if a.phase_artifacts.is_empty() {
ExtractionConsoleState::Planned
} else {
ExtractionConsoleState::Preparing
}
}
fn planned_timeline(plan: Option<&Value>) -> Vec<ExtractionConsoleTimelineEntry> {
let plan_artifact_id = plan.map(artifact_id).unwrap_or_else(|| "plan".into());
plan.map(|v| array(v, "phases"))
.unwrap_or_default()
.into_iter()
.map(|v| ExtractionConsoleTimelineEntry {
phase_id: text(v, "phaseId").unwrap_or_else(|| "unknown-phase".into()),
kind: text(v, "kind").unwrap_or_else(|| "unknown".into()),
state: "planned".into(),
artifact_id: plan_artifact_id.clone(),
blockers: vec![],
evidence: vec![],
})
.collect()
}
fn blockers_from(
value: Option<&Value>,
artifact_id: &str,
field: &str,
) -> Vec<ExtractionConsoleBlocker> {
value
.map(|v| array(v, field))
.unwrap_or_default()
.into_iter()
.map(|v| ExtractionConsoleBlocker {
code: text(v, "code")
.or_else(|| text(v, "issueCode"))
.unwrap_or_else(|| "blocked".into()),
subject: text(v, "subject").unwrap_or_else(|| artifact_id.into()),
detail: text(v, "detail")
.or_else(|| text(v, "message"))
.unwrap_or_else(|| "Extraction phase is blocked.".into()),
next_actions: texts(v, "nextActions"),
artifact_id: artifact_id.into(),
})
.collect()
}
fn evidence_from(value: &Value, artifact_id: &str) -> Vec<ExtractionConsoleEvidence> {
array(value, "evidence")
.into_iter()
.map(|v| ExtractionConsoleEvidence {
kind: text(v, "kind").unwrap_or_else(|| "evidence".into()),
subject: text(v, "subject").unwrap_or_else(|| artifact_id.into()),
digest: text(v, "digest").unwrap_or_else(|| digest(v)),
detail: text(v, "detail").unwrap_or_default(),
artifact_id: artifact_id.into(),
})
.collect()
}
fn approvals(plan: Option<&Value>) -> Vec<ExtractionConsoleApprovalBoundary> {
let mut out = plan
.map(|v| array(v, "phases"))
.unwrap_or_default()
.into_iter()
.filter_map(|phase| {
let b = phase.get("approvalBoundary")?;
Some(ExtractionConsoleApprovalBoundary {
boundary_id: text(b, "boundaryId")?,
phase_id: text(b, "phaseId")
.or_else(|| text(phase, "phaseId"))
.unwrap_or_default(),
action: text(b, "action").unwrap_or_default(),
reason: text(b, "reason").unwrap_or_default(),
required_pins: texts(b, "requiredPins"),
})
})
.collect::<Vec<_>>();
out.sort_by(|l, r| l.boundary_id.cmp(&r.boundary_id));
out
}
fn artifact_id(v: &Value) -> String {
[
"commitId",
"evidenceId",
"cutoverId",
"quiescenceId",
"verificationId",
"reconciliationId",
"runId",
"planId",
]
.into_iter()
.find_map(|f| text(v, f))
.unwrap_or_else(|| digest(v))
}
fn protocol(v: &Value) -> &str {
v.get("protocol")
.and_then(Value::as_str)
.unwrap_or("unknown")
}
fn status(v: &Value) -> &str {
v.get("status")
.and_then(Value::as_str)
.unwrap_or("recorded")
}
fn phase_for(p: &str) -> &str {
match p {
"lenso.extraction-run.v1" => "03-destination-expansion",
"lenso.extraction-backfill.v1" => "04-backfill",
"lenso.extraction-reconciliation.v1" => "05-reconciliation",
"lenso.extraction-verification.v1" => "06-verification",
"lenso.extraction-quiescence.v1" => "07-drain",
"lenso.extraction-provisional-cutover.v1" => "08-provisional-cutover",
"lenso.extraction-authority-commit.v1" => "09-rollback-or-commit",
"lenso.extraction-candidate-health.v1" => "09-rollback-or-commit",
_ => "unknown-phase",
}
}
fn phase_order(v: &str) -> u16 {
v.split('-')
.next()
.and_then(|x| x.parse().ok())
.unwrap_or(u16::MAX)
}
fn text(v: &Value, f: &str) -> Option<String> {
v.get(f).and_then(Value::as_str).map(str::to_owned)
}
fn texts(v: &Value, f: &str) -> Vec<String> {
array(v, f)
.into_iter()
.filter_map(Value::as_str)
.map(str::to_owned)
.collect()
}
fn array<'a>(v: &'a Value, f: &str) -> Vec<&'a Value> {
v.get(f)
.and_then(Value::as_array)
.map(|x| x.iter().collect())
.unwrap_or_default()
}
fn digest(value: &impl Serialize) -> String {
extraction_input_digest(&serde_json::to_vec(value).expect("projection values serialize"))
}