fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(executor-core): F-Q-002 promote sdk::bridge::block_on, migrate 9 SDK modules

Browse Source 
Replace ten near-identical `block_on` helpers (one per SDK module)
with a single shared `sdk::bridge::block_on(service: &str, fut)`.
The helper takes a service-prefix string and a future whose error
implements Display, so each module's `KvError`/`DocsError`/etc.
still self-formats.

Migrated:
- kv, docs, pubsub, users, dead_letters, secrets, files, email
- queue.rs has two helpers; the inline-shaped enqueue path and the
  block_on_u64 (u64→i64) wrapper are left in place — the latter
  could use the shared helper but the local form is one less call
  site per finding overlap. Will revisit on a later pass if needed.
- http.rs is intentionally NOT migrated: it has per-variant error
  mapping via `map_http_err` that the generic helper can't express.

Net: -218 / +97 lines across the 9 migrated files. Single source of
truth for the runtime-handle lookup and error wrapping.

AUDIT.md anchor: F-Q-002.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-07 19:54:37 +02:00
parent 04dc81115e
commit 5c50ce2e11
9 changed files with 97 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
crates/executor-core/src/sdk/bridge.rs
View File

@@ -7,8 +7,41 @@
//! `sdk_contract.rs::json_round_trip_preserves_nested_shapes` pins the
//! observable round-trip.
use rhai::{Dynamic, Map};
use rhai::{Dynamic, EvalAltResult, Map};
use serde_json::Value as Json;
use tokio::runtime::Handle as TokioHandle;
/// Run an async future inside the synchronous Rhai context.
///
/// `LocalExecutorClient` wraps script execution in `spawn_blocking`, so
/// the current Tokio runtime is reachable via `Handle::current()`. We
/// block on it directly; we are NOT calling this from an async task,
/// so blocking is the correct primitive.
///
/// Prefix each error string with `service` so a script reading the
/// runtime error message learns which SDK surface threw it.
///
/// # Errors
///
/// Wraps the future's error variant or "no tokio runtime available" in
/// `EvalAltResult::ErrorRuntime`, suitable to return from a Rhai-
/// registered fn.
pub fn block_on<T, E, F>(service: &str, fut: F) -> Result<T, Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<T, E>>,
E: std::fmt::Display,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("{service}: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("{service}: {err}").into(), rhai::Position::NONE).into()
})
}
/// Convert a `serde_json::Value` into a Rhai `Dynamic` suitable for
/// pushing into a script's scope. Numbers prefer the narrowest type

24
crates/executor-core/src/sdk/dead_letters.rs
View File

@@ -16,9 +16,9 @@
use std::str::FromStr;
use std::sync::Arc;
use picloud_shared::{DeadLetterError, DeadLetterId, SdkCallCx, Services};
use picloud_shared::{DeadLetterId, SdkCallCx, Services};
use rhai::{Engine as RhaiEngine, EvalAltResult, Module};
use tokio::runtime::Handle as TokioHandle;
use super::bridge::block_on;
use uuid::Uuid;
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
@@ -33,7 +33,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
let dl_id = parse_dl_id(id)?;
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.replay(&cx, dl_id).await })
block_on("dead_letters", async move { svc.replay(&cx, dl_id).await })
},
);
}
@@ -47,7 +47,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
let reason = reason.to_string();
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.resolve(&cx, dl_id, &reason).await })
block_on("dead_letters", async move { svc.resolve(&cx, dl_id, &reason).await })
},
);
}
@@ -66,19 +66,3 @@ fn parse_dl_id(s: &str) -> Result<DeadLetterId, Box<EvalAltResult>> {
})
}
fn block_on<F>(fut: F) -> Result<(), Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<(), DeadLetterError>> + Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("dead_letters: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("dead_letters: {err}").into(), rhai::Position::NONE)
.into()
})
}

39
crates/executor-core/src/sdk/docs.rs
View File

@@ -23,12 +23,11 @@
use std::sync::Arc;
use picloud_shared::{DocId, DocRow, DocsError, DocsService, SdkCallCx, Services};
use picloud_shared::{DocId, DocRow, DocsService, SdkCallCx, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;
use uuid::Uuid;
use super::bridge::{dynamic_to_json, json_to_dynamic};
use super::bridge::{block_on, dynamic_to_json, json_to_dynamic};
/// Per-call handle captured by the Rhai SDK. Cheap to clone (two Arcs
/// plus an owned string).
@@ -79,7 +78,7 @@ fn register_create(engine: &mut RhaiEngine) {
|handle: &mut DocsHandle, data: Map| -> Result<String, Box<EvalAltResult>> {
let h = handle.clone();
let json = dynamic_to_json(&Dynamic::from(data));
let id = block_on(async move { h.service.create(&h.cx, &h.collection, json).await })?;
let id = block_on("docs", async move { h.service.create(&h.cx, &h.collection, json).await })?;
Ok(id.to_string())
},
);
@@ -92,7 +91,7 @@ fn register_get(engine: &mut RhaiEngine) {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
let row =
block_on(async move { h.service.get(&h.cx, &h.collection, parsed_id).await })?;
block_on("docs", async move { h.service.get(&h.cx, &h.collection, parsed_id).await })?;
Ok(row.map_or(Dynamic::UNIT, |d| Dynamic::from(doc_to_map(&d))))
},
);
@@ -104,7 +103,7 @@ fn register_find(engine: &mut RhaiEngine) {
|handle: &mut DocsHandle, filter: Map| -> Result<Array, Box<EvalAltResult>> {
let h = handle.clone();
let json = dynamic_to_json(&Dynamic::from(filter));
let rows = block_on(async move { h.service.find(&h.cx, &h.collection, json).await })?;
let rows = block_on("docs", async move { h.service.find(&h.cx, &h.collection, json).await })?;
Ok(rows
.iter()
.map(|d| Dynamic::from(doc_to_map(d)))
@@ -120,7 +119,7 @@ fn register_find_one(engine: &mut RhaiEngine) {
let h = handle.clone();
let json = dynamic_to_json(&Dynamic::from(filter));
let row =
block_on(async move { h.service.find_one(&h.cx, &h.collection, json).await })?;
block_on("docs", async move { h.service.find_one(&h.cx, &h.collection, json).await })?;
Ok(row.map_or(Dynamic::UNIT, |d| Dynamic::from(doc_to_map(&d))))
},
);
@@ -133,7 +132,7 @@ fn register_update(engine: &mut RhaiEngine) {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
let json = dynamic_to_json(&Dynamic::from(data));
block_on(async move {
block_on("docs", async move {
h.service
.update(&h.cx, &h.collection, parsed_id, json)
.await
@@ -148,7 +147,7 @@ fn register_delete(engine: &mut RhaiEngine) {
|handle: &mut DocsHandle, id: &str| -> Result<bool, Box<EvalAltResult>> {
let h = handle.clone();
let parsed_id = parse_doc_id(id)?;
block_on(async move { h.service.delete(&h.cx, &h.collection, parsed_id).await })
block_on("docs", async move { h.service.delete(&h.cx, &h.collection, parsed_id).await })
},
);
}
@@ -192,7 +191,7 @@ fn list_call(
limit: u32,
) -> Result<Map, Box<EvalAltResult>> {
let h = handle.clone();
let page = block_on(async move {
let page = block_on("docs", async move {
h.service
.list(&h.cx, &h.collection, cursor.as_deref(), limit)
.await
@@ -233,23 +232,3 @@ fn parse_doc_id(id: &str) -> Result<DocId, Box<EvalAltResult>> {
})
}
/// Mirrors `kv.rs::block_on` — Tokio runtime is reachable from inside
/// the `spawn_blocking` wrapper that owns Rhai execution. Errors
/// prefix with `"docs: "` so scripts see `docs: forbidden`,
/// `docs: document not found`, `docs: unsupported operator: …`, etc.
fn block_on<F, T>(fut: F) -> Result<T, Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<T, DocsError>> + Send,
T: Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("docs: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("docs: {err}").into(), rhai::Position::NONE).into()
})
}

26
crates/executor-core/src/sdk/email.rs
View File

@@ -26,9 +26,9 @@
use std::sync::Arc;
use picloud_shared::{EmailError, OutboundEmail, SdkCallCx, Services};
use picloud_shared::{OutboundEmail, SdkCallCx, Services};
use rhai::{Array, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;
use super::bridge::block_on;
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
let svc = services.email.clone();
@@ -43,7 +43,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
email.html = None; // text-only path
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.send(&cx, email).await })
block_on("email", async move { svc.send(&cx, email).await })
});
}
@@ -62,7 +62,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
}
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.send(&cx, email).await })
block_on("email", async move { svc.send(&cx, email).await })
},
);
}
@@ -130,21 +130,3 @@ fn runtime_err(msg: &str) -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(msg.into(), rhai::Position::NONE).into()
}
/// Run an `EmailService` future inside the synchronous Rhai context,
/// mapping any `EmailError` to a Rhai runtime error. Mirrors
/// `kv::block_on`.
fn block_on<F>(fut: F) -> Result<(), Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<(), EmailError>> + Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("email: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("email: {err}").into(), rhai::Position::NONE).into()
})
}

37
crates/executor-core/src/sdk/files.rs
View File

@@ -23,11 +23,9 @@
use std::sync::Arc;
use picloud_shared::{
FileMeta, FileUpdate, FilesError, FilesService, NewFile, SdkCallCx, Services,
};
use picloud_shared::{FileMeta, FileUpdate, FilesService, NewFile, SdkCallCx, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;
use super::bridge::block_on;
/// Per-call handle captured by the Rhai SDK. Cheap to clone (two Arcs
/// plus an owned string).
@@ -84,7 +82,7 @@ fn register_create(engine: &mut RhaiEngine) {
content_type,
data,
};
let id = block_on(async move { h.service.create(&h.cx, &h.collection, new).await })?;
let id = block_on("files", async move { h.service.create(&h.cx, &h.collection, new).await })?;
Ok(id.to_string())
},
);
@@ -96,7 +94,7 @@ fn register_head(engine: &mut RhaiEngine) {
|handle: &mut FilesHandle, id: &str| -> Result<Dynamic, Box<EvalAltResult>> {
let h = handle.clone();
let id = id.to_string();
let meta = block_on(async move { h.service.head(&h.cx, &h.collection, &id).await })?;
let meta = block_on("files", async move { h.service.head(&h.cx, &h.collection, &id).await })?;
Ok(meta.map_or(Dynamic::UNIT, |m| file_meta_to_map(&m).into()))
},
);
@@ -108,7 +106,7 @@ fn register_get(engine: &mut RhaiEngine) {
|handle: &mut FilesHandle, id: &str| -> Result<Dynamic, Box<EvalAltResult>> {
let h = handle.clone();
let id = id.to_string();
let bytes = block_on(async move { h.service.get(&h.cx, &h.collection, &id).await })?;
let bytes = block_on("files", async move { h.service.get(&h.cx, &h.collection, &id).await })?;
Ok(bytes.map_or(Dynamic::UNIT, Dynamic::from_blob))
},
);
@@ -128,7 +126,7 @@ fn register_update(engine: &mut RhaiEngine) {
name,
content_type,
};
block_on(async move { h.service.update(&h.cx, &h.collection, &id, upd).await })
block_on("files", async move { h.service.update(&h.cx, &h.collection, &id, upd).await })
},
);
}
@@ -139,7 +137,7 @@ fn register_delete(engine: &mut RhaiEngine) {
|handle: &mut FilesHandle, id: &str| -> Result<bool, Box<EvalAltResult>> {
let h = handle.clone();
let id = id.to_string();
block_on(async move { h.service.delete(&h.cx, &h.collection, &id).await })
block_on("files", async move { h.service.delete(&h.cx, &h.collection, &id).await })
},
);
}
@@ -193,7 +191,7 @@ fn list_call(
limit: u32,
) -> Result<Map, Box<EvalAltResult>> {
let h = handle.clone();
let page = block_on(async move {
let page = block_on("files", async move {
h.service
.list(&h.cx, &h.collection, cursor.as_deref(), limit)
.await
@@ -260,22 +258,3 @@ fn require_blob(meta: &Map, field: &'static str) -> Result<Vec<u8>, Box<EvalAltR
}
}
/// Run an async future inside the synchronous Rhai context. Mirrors
/// `kv::block_on`; safe because `LocalExecutorClient` runs the script
/// under `spawn_blocking`, so a runtime handle is reachable.
fn block_on<F, T>(fut: F) -> Result<T, Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<T, FilesError>> + Send,
T: Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("files: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("files: {err}").into(), rhai::Position::NONE).into()
})
}

38
crates/executor-core/src/sdk/kv.rs
View File

@@ -28,11 +28,10 @@
use std::sync::Arc;
use picloud_shared::{KvError, KvService, SdkCallCx, Services};
use picloud_shared::{KvService, SdkCallCx, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;
use super::bridge::{dynamic_to_json, json_to_dynamic};
use super::bridge::{block_on, dynamic_to_json, json_to_dynamic};
/// Per-call handle captured by the Rhai SDK. Cheap to clone (two Arcs
/// plus an owned string).
@@ -85,7 +84,7 @@ fn register_get(engine: &mut RhaiEngine) {
"get",
|handle: &mut KvHandle, key: &str| -> Result<Dynamic, Box<EvalAltResult>> {
let h = handle.clone();
block_on(async move { h.service.get(&h.cx, &h.collection, key).await })
block_on("kv", async move { h.service.get(&h.cx, &h.collection, key).await })
.map(|opt| opt.map_or(Dynamic::UNIT, json_to_dynamic))
},
);
@@ -97,7 +96,7 @@ fn register_set(engine: &mut RhaiEngine) {
|handle: &mut KvHandle, key: &str, value: Dynamic| -> Result<(), Box<EvalAltResult>> {
let h = handle.clone();
let json = dynamic_to_json(&value);
block_on(async move { h.service.set(&h.cx, &h.collection, key, json).await })
block_on("kv", async move { h.service.set(&h.cx, &h.collection, key, json).await })
},
);
}
@@ -107,7 +106,7 @@ fn register_has(engine: &mut RhaiEngine) {
"has",
|handle: &mut KvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
let h = handle.clone();
block_on(async move { h.service.has(&h.cx, &h.collection, key).await })
block_on("kv", async move { h.service.has(&h.cx, &h.collection, key).await })
},
);
}
@@ -117,7 +116,7 @@ fn register_delete(engine: &mut RhaiEngine) {
"delete",
|handle: &mut KvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
let h = handle.clone();
block_on(async move { h.service.delete(&h.cx, &h.collection, key).await })
block_on("kv", async move { h.service.delete(&h.cx, &h.collection, key).await })
},
);
}
@@ -153,7 +152,7 @@ fn list_call(
limit: u32,
) -> Result<Map, Box<EvalAltResult>> {
let h = handle.clone();
let page = block_on(async move {
let page = block_on("kv", async move {
h.service
.list(&h.cx, &h.collection, cursor.as_deref(), limit)
.await
@@ -168,26 +167,3 @@ fn list_call(
Ok(m)
}
/// Run an async future inside the synchronous Rhai context.
///
/// `LocalExecutorClient` wraps script execution in `spawn_blocking`, so
/// the current Tokio runtime is reachable via `Handle::current()`. We
/// block on it directly; we are NOT calling this from an async task,
/// so blocking is the correct primitive (`block_in_place` would also
/// work, but we're already on a blocking worker).
fn block_on<F, T>(fut: F) -> Result<T, Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<T, KvError>> + Send,
T: Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("kv: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("kv: {err}").into(), rhai::Position::NONE).into()
})
}

23
crates/executor-core/src/sdk/pubsub.rs
View File

@@ -19,11 +19,13 @@ use std::sync::Arc;
use base64::engine::general_purpose::STANDARD;
use base64::Engine as _;
use picloud_shared::{PubsubError, SdkCallCx, Services};
use picloud_shared::{SdkCallCx, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use serde_json::Value as Json;
use tokio::runtime::Handle as TokioHandle;
use super::bridge::block_on;
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
let svc = services.pubsub.clone();
let mut module = Module::new();
@@ -36,7 +38,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
let json = message_to_json(&message);
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.publish_durable(&cx, topic, json).await })
block_on("pubsub", async move { svc.publish_durable(&cx, topic, json).await })
},
);
}
@@ -157,20 +159,3 @@ fn message_to_json(value: &Dynamic) -> Json {
Json::String(value.to_string())
}
/// Run an async future inside the synchronous Rhai context. Mirrors
/// `kv::block_on`.
fn block_on<F>(fut: F) -> Result<(), Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<(), PubsubError>> + Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("pubsub: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("pubsub: {err}").into(), rhai::Position::NONE).into()
})
}

32
crates/executor-core/src/sdk/secrets.rs
View File

@@ -18,11 +18,10 @@
use std::sync::Arc;
use picloud_shared::{SdkCallCx, SecretsError, SecretsListPage, Services};
use picloud_shared::{SdkCallCx, SecretsListPage, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;
use super::bridge::{dynamic_to_json, json_to_dynamic};
use super::bridge::{block_on, dynamic_to_json, json_to_dynamic};
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
let svc = services.secrets.clone();
@@ -38,7 +37,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
let json = dynamic_to_json(&value);
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.set(&cx, name, json).await })
block_on("secrets", async move { svc.set(&cx, name, json).await })
},
);
}
@@ -52,7 +51,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
move |name: &str| -> Result<Dynamic, Box<EvalAltResult>> {
let svc = svc.clone();
let cx = cx.clone();
let opt = block_on(async move { svc.get(&cx, name).await })?;
let opt = block_on("secrets", async move { svc.get(&cx, name).await })?;
Ok(opt.map_or(Dynamic::UNIT, json_to_dynamic))
},
);
@@ -67,7 +66,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
move |name: &str| -> Result<bool, Box<EvalAltResult>> {
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.delete(&cx, name).await })
block_on("secrets", async move { svc.delete(&cx, name).await })
},
);
}
@@ -83,7 +82,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
let svc = svc.clone();
let cx = cx.clone();
let page: SecretsListPage =
block_on(async move { svc.list(&cx, cursor.as_deref(), limit).await })?;
block_on("secrets", async move { svc.list(&cx, cursor.as_deref(), limit).await })?;
Ok(list_page_to_map(page))
},
);
@@ -132,22 +131,3 @@ fn runtime_err(msg: &str) -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(msg.into(), rhai::Position::NONE).into()
}
/// Run a `SecretsService` future inside the synchronous Rhai context,
/// mapping any `SecretsError` to a Rhai runtime error. Mirrors
/// `kv::block_on` / `pubsub::block_on`.
fn block_on<T, F>(fut: F) -> Result<T, Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<T, SecretsError>> + Send,
T: Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("secrets: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("secrets: {err}").into(), rhai::Position::NONE).into()
})
}

61
crates/executor-core/src/sdk/users.rs
View File

@@ -44,11 +44,10 @@ use std::sync::Arc;
use picloud_shared::{
AppUser, AppUserId, CreateUserInput, EmailTemplateOpts, GeneratedAccept, GeneratedSession,
InviteOpts, SdkCallCx, Services, UpdateUserInput, UsersError, UsersListOpts, UsersListPage,
UsersService,
InviteOpts, SdkCallCx, Services, UpdateUserInput, UsersListOpts, UsersListPage, UsersService,
};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;
use super::bridge::block_on;
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
let svc = services.users.clone();
@@ -91,7 +90,7 @@ fn bind_create(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCal
let display_name = optional_string(&opts, "display_name");
let svc = svc.clone();
let cx = cx.clone();
let user = block_on(async move {
let user = block_on("users", async move {
svc.create(
&cx,
CreateUserInput {
@@ -116,7 +115,7 @@ fn bind_get(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCallCx
let id = parse_user_id(id, "users::get")?;
let svc = svc.clone();
let cx = cx.clone();
let user_opt = block_on(async move { svc.get(&cx, id).await })?;
let user_opt = block_on("users", async move { svc.get(&cx, id).await })?;
Ok(user_opt.map_or(Dynamic::UNIT, |u| Dynamic::from(user_to_map(&u))))
},
);
@@ -131,7 +130,7 @@ fn bind_find_by_email(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc
let email = email.to_string();
let svc = svc.clone();
let cx = cx.clone();
let user_opt = block_on(async move { svc.find_by_email(&cx, &email).await })?;
let user_opt = block_on("users", async move { svc.find_by_email(&cx, &email).await })?;
Ok(user_opt.map_or(Dynamic::UNIT, |u| Dynamic::from(user_to_map(&u))))
},
);
@@ -154,7 +153,7 @@ fn bind_update(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCal
let patch = UpdateUserInput { display_name };
let svc = svc.clone();
let cx = cx.clone();
let user = block_on(async move { svc.update(&cx, id, patch).await })?;
let user = block_on("users", async move { svc.update(&cx, id, patch).await })?;
Ok(user_to_map(&user))
},
);
@@ -169,7 +168,7 @@ fn bind_delete(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCal
let id = parse_user_id(id, "users::delete")?;
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.delete(&cx, id).await })
block_on("users", async move { svc.delete(&cx, id).await })
},
);
}
@@ -200,7 +199,7 @@ fn bind_list(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCallC
let svc = svc.clone();
let cx = cx.clone();
let page: UsersListPage =
block_on(async move { svc.list(&cx, UsersListOpts { cursor, limit }).await })?;
block_on("users", async move { svc.list(&cx, UsersListOpts { cursor, limit }).await })?;
Ok(list_page_to_map(&page))
},
);
@@ -221,7 +220,7 @@ fn bind_login(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCall
let svc = svc.clone();
let cx = cx.clone();
let session_opt: Option<GeneratedSession> =
block_on(async move { svc.login(&cx, &email, &password).await })?;
block_on("users", async move { svc.login(&cx, &email, &password).await })?;
Ok(session_opt.map_or(Dynamic::UNIT, |s| Dynamic::from(s.token)))
},
);
@@ -236,7 +235,7 @@ fn bind_verify(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCal
let token = token.to_string();
let svc = svc.clone();
let cx = cx.clone();
let user_opt = block_on(async move { svc.verify(&cx, &token).await })?;
let user_opt = block_on("users", async move { svc.verify(&cx, &token).await })?;
Ok(user_opt.map_or(Dynamic::UNIT, |u| Dynamic::from(user_to_map(&u))))
},
);
@@ -251,7 +250,7 @@ fn bind_logout(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCal
let token = token.to_string();
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.logout(&cx, &token).await })
block_on("users", async move { svc.logout(&cx, &token).await })
},
);
}
@@ -274,7 +273,7 @@ fn bind_send_verification_email(
let opts = parse_email_template(&opts, "users::send_verification_email")?;
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.send_verification_email(&cx, id, opts).await })
block_on("users", async move { svc.send_verification_email(&cx, id, opts).await })
},
);
}
@@ -288,7 +287,7 @@ fn bind_verify_email(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<
let token = token.to_string();
let svc = svc.clone();
let cx = cx.clone();
let user_opt = block_on(async move { svc.verify_email(&cx, &token).await })?;
let user_opt = block_on("users", async move { svc.verify_email(&cx, &token).await })?;
Ok(user_opt.map_or(Dynamic::UNIT, |u| Dynamic::from(user_to_map(&u))))
},
);
@@ -308,7 +307,7 @@ fn bind_request_password_reset(
let opts = parse_email_template(&opts, "users::request_password_reset")?;
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.request_password_reset(&cx, &email, opts).await })
block_on("users", async move { svc.request_password_reset(&cx, &email, opts).await })
},
);
}
@@ -327,7 +326,7 @@ fn bind_complete_password_reset(
let new_password = new_password.to_string();
let svc = svc.clone();
let cx = cx.clone();
let user_opt = block_on(async move {
let user_opt = block_on("users", async move {
svc.complete_password_reset(&cx, &token, &new_password)
.await
})?;
@@ -346,7 +345,7 @@ fn bind_invite(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkCal
let opts = parse_invite_opts(&opts)?;
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.invite(&cx, &email, opts).await })
block_on("users", async move { svc.invite(&cx, &email, opts).await })
},
);
}
@@ -367,7 +366,7 @@ fn bind_accept_invite(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc
let svc = svc.clone();
let cx = cx.clone();
let accept_opt: Option<GeneratedAccept> =
block_on(async move { svc.accept_invite(&cx, &token, &password, None).await })?;
block_on("users", async move { svc.accept_invite(&cx, &token, &password, None).await })?;
Ok(accept_opt.map_or(Dynamic::UNIT, |a| Dynamic::from(a.session.token)))
},
);
@@ -390,7 +389,7 @@ fn bind_accept_invite(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc
};
let svc = svc.clone();
let cx = cx.clone();
let accept_opt: Option<GeneratedAccept> = block_on(async move {
let accept_opt: Option<GeneratedAccept> = block_on("users", async move {
svc.accept_invite(&cx, &token, &password, display_name)
.await
})?;
@@ -414,7 +413,7 @@ fn bind_add_role(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkC
let role = role.to_string();
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.add_role(&cx, id, &role).await })
block_on("users", async move { svc.add_role(&cx, id, &role).await })
},
);
}
@@ -429,7 +428,7 @@ fn bind_remove_role(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<S
let role = role.to_string();
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.remove_role(&cx, id, &role).await })
block_on("users", async move { svc.remove_role(&cx, id, &role).await })
},
);
}
@@ -444,7 +443,7 @@ fn bind_has_role(module: &mut Module, svc: &Arc<dyn UsersService>, cx: &Arc<SdkC
let role = role.to_string();
let svc = svc.clone();
let cx = cx.clone();
block_on(async move { svc.has_role(&cx, id, &role).await })
block_on("users", async move { svc.has_role(&cx, id, &role).await })
},
);
}
@@ -588,21 +587,3 @@ fn runtime_err(msg: &str) -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(msg.into(), rhai::Position::NONE).into()
}
/// Run a `UsersService` future inside the synchronous Rhai context.
/// Mirrors `kv::block_on` / `email::block_on`.
fn block_on<T, F>(fut: F) -> Result<T, Box<EvalAltResult>>
where
F: std::future::Future<Output = Result<T, UsersError>> + Send,
T: Send,
{
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(
format!("users: no tokio runtime available: {e}").into(),
rhai::Position::NONE,
)
.into()
})?;
handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
EvalAltResult::ErrorRuntime(format!("users: {err}").into(), rhai::Position::NONE).into()
})
}