use super::*;
impl Store {
pub async fn stage_orientation_candidate(
&self,
session: &SessionId,
candidate_id: &str,
scope: &OrientationScope,
candidate_semantic: Value,
candidate_json: Value,
) -> Result<Value> {
ensure!(!candidate_id.is_empty(), "invalid_orientation_identity");
ensure!(
candidate_json["candidate_id"].as_str() == Some(candidate_id),
"candidate_identity_mismatch"
);
ensure!(
candidate_json["scope"] == serde_json::to_value(scope)?,
"orientation_scope_mismatch"
);
let mut semantic = candidate_json.clone();
semantic
.as_object_mut()
.context("invalid_orientation_candidate")?
.remove("staged_at_ms");
ensure!(
semantic == candidate_semantic,
"orientation_candidate_semantic_mismatch"
);
let (session, candidate_id, scope) =
(session.to_string(), candidate_id.to_owned(), scope.label());
self.with(move |db| {
let tx = db.transaction_with_behavior(TransactionBehavior::Immediate)?;
let prior: Option<(String, String, String)> = tx.query_row(
"SELECT scope,semantic,candidate FROM orientation_candidates WHERE session_id=?1 AND candidate_id=?2",
params![session, candidate_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
).optional()?;
if let Some((prior_scope, semantic, candidate)) = prior {
ensure!(prior_scope == scope && serde_json::from_str::<Value>(&semantic)? == candidate_semantic,
"orientation_candidate_payload_conflict");
return Ok(serde_json::from_str(&candidate)?);
}
tx.execute(
"INSERT INTO orientation_candidates(session_id,candidate_id,scope,semantic,candidate) VALUES(?1,?2,?3,?4,?5)",
params![session,candidate_id,scope,candidate_semantic.to_string(),candidate_json.to_string()],
)?;
tx.commit()?;
Ok(candidate_json)
}).await
}
pub async fn orientation_candidate(
&self,
session: &SessionId,
candidate_id: &str,
) -> Result<Option<Value>> {
let (session, candidate_id) = (session.to_string(), candidate_id.to_owned());
self.with(move |db| {
let raw: Option<String> = db.query_row(
"SELECT candidate FROM orientation_candidates WHERE session_id=?1 AND candidate_id=?2",
params![session,candidate_id], |r| r.get(0),
).optional()?;
raw.map(|s| serde_json::from_str(&s).map_err(Into::into)).transpose()
}).await
}
pub async fn current_orientation_revision(
&self,
session: &SessionId,
scope: &OrientationScope,
) -> Result<Option<Value>> {
let (session, scope) = (session.to_string(), scope.label());
self.with(move |db| {
let raw: Option<String> = db
.query_row(
"SELECT r.revision FROM orientation_selections s JOIN orientation_revisions r
ON r.session_id=s.session_id AND r.scope=s.scope AND r.revision_id=s.revision_id
WHERE s.session_id=?1 AND s.scope=?2",
params![session, scope],
|r| r.get(0),
)
.optional()?;
raw.map(|s| serde_json::from_str(&s).map_err(Into::into))
.transpose()
})
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn commit_orientation_adoption(
&self,
session: &SessionId,
scope: &OrientationScope,
candidate_id: &str,
expected_parent: Option<&str>,
operation_id: &str,
canonical_request: Value,
revision_json: Value,
receipt_json: Value,
) -> Result<Event> {
let (session, scope, candidate_id, operation_id) = (
session.to_string(),
scope.label(),
candidate_id.to_owned(),
operation_id.to_owned(),
);
let expected_parent = expected_parent.map(str::to_owned);
self.with(move |db| {
let tx = db.transaction_with_behavior(TransactionBehavior::Immediate)?;
let prior: Option<(String, String, i64)> = tx.query_row(
"SELECT scope,request,event_id FROM orientation_revisions WHERE session_id=?1 AND operation_id=?2",
params![session,operation_id], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?)),
).optional()?;
if let Some((prior_scope, request, event_id)) = prior {
ensure!(prior_scope == scope && serde_json::from_str::<Value>(&request)? == canonical_request,
"orientation_operation_payload_conflict");
return orientation_event(&tx, event_id);
}
let candidate: Option<(String, String)> = tx.query_row(
"SELECT scope,semantic FROM orientation_candidates WHERE session_id=?1 AND candidate_id=?2",
params![session,candidate_id], |r| Ok((r.get(0)?,r.get(1)?)),
).optional()?;
let (candidate_scope, semantic) = candidate.context("orientation_candidate_not_found")?;
ensure!(candidate_scope == scope, "orientation_scope_mismatch");
let semantic: Value = serde_json::from_str(&semantic)?;
ensure!(semantic == canonical_request["candidate"], "orientation_candidate_payload_conflict");
ensure!(canonical_request["expected_parent"] == json!(expected_parent)
&& semantic["parent_revision"] == json!(expected_parent), "orientation_parent_mismatch");
for (key, value) in semantic.as_object().context("invalid_orientation_candidate")? {
ensure!(revision_json.get(key) == Some(value), "orientation_revision_candidate_mismatch: {key}");
}
for key in ["scope", "carrier_hash", "carrier_role", "carrier_placement"] {
ensure!(receipt_json.get(key) == semantic.get(key), "orientation_receipt_candidate_mismatch: {key}");
}
ensure!(revision_json["adoption_actor"] == canonical_request["adoption_actor"]
&& receipt_json["apply_at"] == canonical_request["apply_at"], "orientation_adoption_payload_mismatch");
let current: Option<String> = tx.query_row(
"SELECT revision_id FROM orientation_selections WHERE session_id=?1 AND scope=?2",
params![session,scope], |r| r.get(0),
).optional()?;
ensure!(current == expected_parent, "orientation_stale_parent_conflict");
let revision_id = revision_json["revision_id"].as_str().filter(|s| !s.is_empty())
.context("invalid_orientation_revision")?;
for record in [&revision_json, &receipt_json] {
ensure!(record["candidate_id"].as_str() == Some(candidate_id.as_str())
&& record["operation_id"].as_str() == Some(operation_id.as_str())
&& record["revision_id"].as_str() == Some(revision_id), "orientation_record_identity_mismatch");
}
ensure!(revision_json["parent_revision"] == json!(expected_parent), "orientation_parent_mismatch");
let payload = json!({"type":"orientation.adopted", "operation_id":operation_id,
"candidate_id":candidate_id, "revision_id":revision_id,
"revision":revision_json,"receipt":receipt_json});
let event_id = event(&tx, &session, "orientation.adopted", &payload)?;
tx.execute(
"INSERT INTO orientation_revisions(session_id,revision_id,scope,candidate_id,operation_id,request,revision,event_id)
VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",
params![session,revision_id,scope,candidate_id,operation_id,canonical_request.to_string(),revision_json.to_string(),event_id],
)?;
tx.execute(
"INSERT INTO orientation_selections(session_id,scope,revision_id) VALUES(?1,?2,?3)
ON CONFLICT(session_id,scope) DO UPDATE SET revision_id=excluded.revision_id",
params![session,scope,revision_id],
)?;
let committed = orientation_event(&tx, event_id)?;
tx.commit()?;
Ok(committed)
}).await
}
}
fn orientation_event(db: &Connection, id: i64) -> Result<Event> {
let (kind, payload, created_ms): (String, String, i64) = db.query_row(
"SELECT kind,payload,created_ms FROM events WHERE seq=?1",
[id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)?;
Ok(Event {
id,
kind,
payload: serde_json::from_str(&payload)?,
created_ms,
})
}