use crate::workflow::host_exports::latest::obelisk::workflow::workflow_support::StubJsonError;
use assert_matches::assert_matches;
use chrono::DateTime;
use concepts::ExecutionId;
use concepts::FunctionFqn;
use concepts::prefixed_ulid::ExecutionIdDerived;
use concepts::storage::HistoryEventScheduleAt;
use concepts::storage::StubError;
use indexmap::indexmap;
use std::ops::Deref as _;
use std::time::Duration;
use val_json::wast_val::ValKey;
use val_json::wast_val::WastVal;
pub(crate) use concepts::SUFFIX_FN_AWAIT_NEXT;
pub(crate) use concepts::SUFFIX_FN_GET;
pub(crate) use concepts::SUFFIX_FN_SCHEDULE;
pub(crate) use concepts::SUFFIX_FN_STUB;
pub(crate) use concepts::SUFFIX_FN_SUBMIT;
pub(crate) mod latest {
use chrono::DateTime;
use chrono::Utc;
use concepts::ExecutionId;
use concepts::FunctionFqn;
use concepts::prefixed_ulid::DelayId;
use concepts::prefixed_ulid::ExecutionIdDerived;
use concepts::prefixed_ulid::ExecutionIdParseError;
use concepts::storage::HistoryEventScheduleAt;
use obelisk::types::execution as types_execution;
pub(crate) use obelisk::types::execution::DelayId as DelayIdTypes;
pub(crate) use obelisk::types::execution::ExecutionId as ExecutionIdTypes;
pub(crate) use obelisk::types::execution::ResponseId as ResponseIdTypes;
use obelisk::types::time::Datetime;
pub(crate) use obelisk::types::time::Duration as DurationEnumTypes;
pub(crate) use obelisk::types::time::ScheduleAt as ScheduleAtTypes;
use std::str::FromStr;
use std::time::Duration;
use std::time::UNIX_EPOCH;
wasmtime::component::bindgen!({
path: "host-wit-workflow/",
inline: "package any:any;
world bindings {
import obelisk:workflow/workflow-support@6.0.0;
import obelisk:workflow/workflow-support-backtrace@6.0.0;
}",
world: "any:any/bindings",
with: {
"obelisk:types/join-set.join-set": concepts::JoinSetId,
},
imports: {
default: trappable | async
}
});
impl From<DurationEnumTypes> for Duration {
fn from(value: DurationEnumTypes) -> Self {
match value {
DurationEnumTypes::Milliseconds(millis) => Duration::from_millis(millis),
DurationEnumTypes::Seconds(secs) => Duration::from_secs(secs),
DurationEnumTypes::Minutes(mins) => Duration::from_secs(u64::from(mins * 60)),
DurationEnumTypes::Hours(hours) => Duration::from_secs(u64::from(hours * 60 * 60)),
DurationEnumTypes::Days(days) => {
Duration::from_secs(u64::from(days * 24 * 60 * 60))
}
}
}
}
impl From<Datetime> for DateTime<Utc> {
fn from(
Datetime {
seconds,
nanoseconds,
}: Datetime,
) -> Self {
let duration = Duration::new(seconds, nanoseconds);
let systemtime = UNIX_EPOCH + duration;
DateTime::<Utc>::from(systemtime)
}
}
impl TryFrom<DateTime<Utc>> for Datetime {
type Error = chrono::OutOfRangeError;
fn try_from(value: DateTime<Utc>) -> Result<Self, Self::Error> {
let epoch = DateTime::<Utc>::from(UNIX_EPOCH);
let duration = value.signed_duration_since(epoch).to_std()?;
Ok(Datetime {
seconds: duration.as_secs(),
nanoseconds: duration.subsec_nanos(),
})
}
}
impl From<ScheduleAtTypes> for HistoryEventScheduleAt {
fn from(value: ScheduleAtTypes) -> Self {
match value {
ScheduleAtTypes::Now => Self::Now,
ScheduleAtTypes::At(datetime) => Self::At(DateTime::from(datetime)),
ScheduleAtTypes::In(duration) => Self::In(Duration::from(duration)),
}
}
}
impl From<&FunctionFqn> for types_execution::Function {
fn from(ffqn: &FunctionFqn) -> Self {
Self {
interface_name: ffqn.ifc_fqn.to_string(),
function_name: ffqn.function_name.to_string(),
}
}
}
impl From<&ExecutionIdDerived> for types_execution::ExecutionId {
fn from(value: &ExecutionIdDerived) -> Self {
Self {
id: value.to_string(),
}
}
}
impl From<&ExecutionId> for types_execution::ExecutionId {
fn from(value: &ExecutionId) -> Self {
Self {
id: value.to_string(),
}
}
}
impl TryFrom<types_execution::ExecutionId> for ExecutionId {
type Error = ExecutionIdParseError;
fn try_from(value: types_execution::ExecutionId) -> Result<Self, Self::Error> {
ExecutionId::from_str(&value.id)
}
}
impl From<&DelayId> for types_execution::DelayId {
fn from(value: &DelayId) -> Self {
Self {
id: value.to_string(),
}
}
}
mod into_val {
use crate::workflow::host_exports::latest::ResponseIdTypes;
use super::obelisk::types::execution as types_execution;
use wasmtime::component::Val;
impl From<types_execution::Function> for Val {
fn from(value: types_execution::Function) -> Self {
Self::Record(vec![
(
"interface-name".to_string(),
Self::String(value.interface_name),
),
(
"function-name".to_string(),
Self::String(value.function_name),
),
])
}
}
impl From<types_execution::ExecutionId> for Val {
fn from(value: types_execution::ExecutionId) -> Self {
Self::Record(vec![("id".to_string(), Self::String(value.id))])
}
}
impl From<types_execution::DelayId> for Val {
fn from(value: types_execution::DelayId) -> Self {
Self::Record(vec![("id".to_string(), Self::String(value.id))])
}
}
impl From<ResponseIdTypes> for Val {
fn from(value: ResponseIdTypes) -> Self {
match value {
ResponseIdTypes::ExecutionId(execution_id) => Self::Variant(
"execution-id".to_string(),
Some(Box::new(Self::from(execution_id))),
),
ResponseIdTypes::DelayId(delay_id) => {
Self::Variant("delay-id".to_string(), Some(Box::new(Self::from(delay_id))))
}
}
}
}
impl From<types_execution::FunctionMismatch> for Val {
fn from(value: types_execution::FunctionMismatch) -> Self {
Self::Record(vec![
(
"specified-function".to_string(),
Val::from(value.specified_function),
),
(
"actual-function".to_string(),
Self::Option(value.actual_function.map(|f| Box::new(Self::from(f)))),
),
("actual-id".to_string(), Self::from(value.actual_id)),
])
}
}
impl From<types_execution::GetExtensionError> for Val {
fn from(value: types_execution::GetExtensionError) -> Self {
match value {
types_execution::GetExtensionError::FunctionMismatch(function_mismatch) => {
Self::Variant(
"function-mismatch".to_string(),
Some(Box::new(Val::from(function_mismatch))),
)
}
types_execution::GetExtensionError::NotFoundInProcessedResponses => {
Self::Variant("not-found-in-processed-responses".to_string(), None)
}
}
}
}
}
}
pub(crate) mod response_id {
use db_common::JoinSetResponseId;
use crate::workflow::host_exports::latest::{DelayIdTypes, ExecutionIdTypes, ResponseIdTypes};
impl From<JoinSetResponseId> for ResponseIdTypes {
fn from(value: JoinSetResponseId) -> Self {
match value {
JoinSetResponseId::ChildExecutionId(child_execution_id) => {
ResponseIdTypes::ExecutionId(ExecutionIdTypes::from(&child_execution_id))
}
JoinSetResponseId::DelayId(delay_id) => {
ResponseIdTypes::DelayId(DelayIdTypes::from(&delay_id))
}
}
}
}
}
pub fn history_event_schedule_at_from_wast_val(
scheduled_at: &WastVal,
) -> Result<HistoryEventScheduleAt, &'static str> {
let WastVal::Variant(variant, val) = scheduled_at else {
return Err("wrong type");
};
match (variant.as_snake_str(), val) {
("now", None) => Ok(HistoryEventScheduleAt::Now),
("in", Some(duration)) => {
if let &WastVal::Variant(key, value) = &duration.deref() {
let duration = match (key.as_snake_str(), value.as_deref()) {
("milliseconds", Some(WastVal::U64(value))) => Duration::from_millis(*value),
("seconds", Some(WastVal::U64(value))) => Duration::from_secs(*value),
("minutes", Some(WastVal::U64(value))) => Duration::from_secs(*value * 60),
("hours", Some(WastVal::U64(value))) => Duration::from_secs(*value * 60 * 60),
("days", Some(WastVal::U64(value))) => {
Duration::from_secs(*value * 60 * 60 * 24)
}
_ => {
return Err(
"cannot convert `scheduled-at`, `in` variant: value must be one of the following keys: `milliseconds`(U64), `seconds`(U64), `minutes`(U32), `hours`(U32), `days`(U32)",
);
}
};
Ok(HistoryEventScheduleAt::In(duration))
} else {
Err("cannot convert `scheduled-at`, `in` variant: value must be a variant")
}
}
("at", Some(date_time)) if matches!(date_time.deref(), WastVal::Record(_)) => {
let date_time =
assert_matches!(date_time.deref(), WastVal::Record(keys_vals) => keys_vals)
.iter()
.map(|(k, v)| (k.as_snake_str(), v))
.collect::<std::collections::HashMap<_, _>>();
let seconds = date_time.get("seconds");
let nanoseconds = date_time.get("nanoseconds");
match (date_time.len(), seconds, nanoseconds) {
(2, Some(WastVal::U64(seconds)), Some(WastVal::U32(nanoseconds))) => {
let date_time = latest::obelisk::types::time::Datetime {
seconds: *seconds,
nanoseconds: *nanoseconds,
};
let date_time = DateTime::from(date_time);
Ok(HistoryEventScheduleAt::At(date_time))
}
_ => Err(
"cannot convert `scheduled-at`, `at` variant: record must have exactly two keys: `seconds`(U64), `nanoseconds`(U32)",
),
}
}
_ => Err("cannot convert `scheduled-at` variant, expected one of `now`, `in`, `at`"),
}
}
pub(crate) fn execution_id_into_wast_val(execution_id: &ExecutionId) -> WastVal {
WastVal::Record(
indexmap! {ValKey::new_snake("id") => WastVal::String(execution_id.to_string())},
)
}
pub(crate) fn execution_id_derived_into_wast_val(execution_id: &ExecutionIdDerived) -> WastVal {
WastVal::Record(
indexmap! {ValKey::new_snake("id") => WastVal::String(execution_id.to_string())},
)
}
pub(crate) fn ffqn_into_wast_val(ffqn: &FunctionFqn) -> WastVal {
WastVal::Record(indexmap! {
ValKey::new_snake("interface_name") => WastVal::String(ffqn.ifc_fqn.to_string()),
ValKey::new_snake("function_name") => WastVal::String(ffqn.function_name.to_string()),
})
}
impl From<StubError> for StubJsonError {
fn from(value: StubError) -> StubJsonError {
match value {
StubError::ExecutionNotFound => StubJsonError::ExecutionNotFound,
StubError::TypeCheckError(reason) => StubJsonError::TypeCheckError(reason),
StubError::Conflict => StubJsonError::Conflict,
}
}
}
pub(crate) fn stub_result_to_wast_val(stub_result: Result<(), StubError>) -> WastVal {
match stub_result {
Ok(()) => WastVal::Result(Ok(None)),
Err(err) => {
let (variant, payload) = match err {
StubError::ExecutionNotFound => ("execution-not-found", None),
StubError::Conflict => ("conflict", None),
StubError::TypeCheckError(reason) => {
("type-check-error", Some(Box::new(WastVal::String(reason))))
}
};
WastVal::Result(Err(Some(Box::new(WastVal::Variant(
ValKey::from_kebab(variant),
payload,
)))))
}
}
}