use gwk_domain::envelope::{Actor, ENVELOPE_SCHEMA_VERSION, EventEnvelope, Origin};
use gwk_domain::ids::{
AggregateId, EventId, FenceToken, IdempotencyKey, ProjectId, Seq, Timestamp,
};
use gwk_domain::protocol::{CONTRACT_VERSION, KernelErrorCode};
use sqlx::PgConnection;
use crate::project::Refusal;
use crate::store::{PgEventStore, current_aggregate_version};
pub const KERNEL_AGGREGATE: &str = "kernel";
pub const KERNEL_SINGLETON: &str = "singleton";
pub const SYSTEM_PROJECT: &str = "system";
pub const GENESIS_KEY: &str = "kernel_initialized:v1";
pub const GENESIS_EVENT_TYPE: &str = "kernel_initialized";
pub const ACTIVATION_EVENT_TYPE: &str = "kernel_activated";
const REVISION_HEX_LEN: usize = 40;
const COMMITTED_CUTOVER: &str = "SELECT payload ->> 'cutover_id' FROM gwk.event \
WHERE aggregate_type = 'kernel' AND aggregate_id = 'singleton' \
AND event_type = 'kernel_activated' \
ORDER BY aggregate_version LIMIT 1";
const DB_NOW: &str = "SELECT to_json(now()) #>> '{}'";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Epoch {
None,
Sealed,
Active,
}
impl Epoch {
fn of_version(version: u32) -> Self {
match version {
0 => Self::None,
1 => Self::Sealed,
_ => Self::Active,
}
}
}
pub(crate) async fn epoch_of(conn: &mut PgConnection) -> Result<Epoch, Refusal> {
let version = current_aggregate_version(conn, KERNEL_AGGREGATE, KERNEL_SINGLETON).await?;
Ok(Epoch::of_version(version))
}
pub(crate) async fn committed_cutover(conn: &mut PgConnection) -> Result<Option<String>, Refusal> {
sqlx::query_scalar(COMMITTED_CUTOVER)
.fetch_optional(conn)
.await
.map(Option::flatten)
.map_err(|e| Refusal::storage(format!("read the committed cutover: {e}")))
}
pub(crate) fn activation_key(cutover_id: &str) -> String {
format!("{ACTIVATION_EVENT_TYPE}:{cutover_id}")
}
pub fn is_public_revision(value: &str) -> bool {
value.len() == REVISION_HEX_LEN
&& value
.bytes()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}
impl PgEventStore {
pub async fn ensure_genesis(&self, public_revision: &str) -> Result<(), Refusal> {
if !is_public_revision(public_revision) {
return Err(Refusal::validation(format!(
"genesis records the full 40-character public revision, not {public_revision:?}"
)));
}
let mut tx = self
.pool()
.begin()
.await
.map_err(|e| Refusal::storage(format!("begin: {e}")))?;
let writer = self.lock_writer(&mut tx).await?;
if epoch_of(&mut tx).await? != Epoch::None {
return Ok(());
}
let now: String = sqlx::query_scalar(DB_NOW)
.fetch_one(&mut *tx)
.await
.map_err(|e| Refusal::storage(format!("read the database clock: {e}")))?;
let fence = writer.current_fence.map(FenceToken::new);
self.append_locked(
&mut tx,
&writer,
0,
fence,
&[genesis_event(public_revision, &now)],
)
.await?;
tx.commit()
.await
.map_err(|e| Refusal::storage(format!("commit: {e}")))?;
Ok(())
}
}
fn genesis_event(public_revision: &str, at: &str) -> EventEnvelope {
EventEnvelope {
event_id: EventId::new(format!("{KERNEL_AGGREGATE}:{KERNEL_SINGLETON}:1")),
project_id: ProjectId::new(SYSTEM_PROJECT),
aggregate_type: KERNEL_AGGREGATE.to_owned(),
aggregate_id: AggregateId::new(KERNEL_SINGLETON),
aggregate_version: 1,
event_type: GENESIS_EVENT_TYPE.to_owned(),
schema_version: ENVELOPE_SCHEMA_VERSION,
global_sequence: Seq::new(0),
appended_at: Timestamp::new(at),
occurred_at: Timestamp::new(at),
actor: Actor {
kind: "kernel".to_owned(),
id: None,
},
origin: Origin {
system: "gw".to_owned(),
r#ref: None,
},
causation_id: None,
correlation_id: None,
idempotency_key: Some(IdempotencyKey::new(GENESIS_KEY)),
payload: serde_json::json!({
"contract_version": CONTRACT_VERSION,
"public_revision": public_revision,
}),
payload_ref: None,
}
}
pub(crate) fn sealed_refusal(epoch: Epoch, command_type: &str) -> Refusal {
let message = match epoch {
Epoch::None => {
"this kernel has no epoch: genesis has not been appended, so nothing may be \
submitted — not even activation, which asserts the sealed version"
.to_owned()
}
_ => format!("this kernel is sealed; {command_type} is not on the sealed allowlist"),
};
Refusal::new(KernelErrorCode::Sealed, message)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_version_is_the_epoch() {
assert_eq!(Epoch::of_version(0), Epoch::None);
assert_eq!(Epoch::of_version(1), Epoch::Sealed);
assert_eq!(Epoch::of_version(2), Epoch::Active);
assert_eq!(Epoch::of_version(9), Epoch::Active);
}
#[test]
fn only_a_full_lowercase_revision_is_one() {
assert!(is_public_revision(&"a1b2c3d4e5".repeat(4)));
assert!(!is_public_revision(&"A1B2C3D4E5".repeat(4)));
assert!(!is_public_revision("a1b2c3d4"));
assert!(!is_public_revision(""));
assert!(!is_public_revision(&"z".repeat(40)));
}
#[test]
fn the_activation_key_names_its_cutover() {
assert_eq!(activation_key("c-1"), "kernel_activated:c-1");
}
}