feat: [US-053] - [Inspector: Workflow bridge via NAPI callbacks]

Author: NathanFlurry

rivetkit-rust/packages/rivetkit-core/src/actor/callbacks.rs

@@ -204,6 +204,14 @@ pub type ActionHandler =
 pub type BeforeActionResponseCallback = Box<
 	dyn Fn(OnBeforeActionResponseRequest) -> BoxFuture<'static, Result<Vec<u8>>> + Send + Sync,
 >;
+pub type GetWorkflowHistoryCallback = Box<
+	dyn Fn(GetWorkflowHistoryRequest) -> BoxFuture<'static, Result<Option<Vec<u8>>>>
+		+ Send
+		+ Sync,
+>;
+pub type ReplayWorkflowCallback = Box<
+	dyn Fn(ReplayWorkflowRequest) -> BoxFuture<'static, Result<Option<Vec<u8>>>> + Send + Sync,
+>;
 
 #[derive(Clone, Debug)]
 pub struct OnWakeRequest {
@@ -283,6 +291,17 @@ pub struct RunRequest {
 	pub ctx: ActorContext,
 }
 
+#[derive(Clone, Debug)]
+pub struct GetWorkflowHistoryRequest {
+	pub ctx: ActorContext,
+}
+
+#[derive(Clone, Debug)]
+pub struct ReplayWorkflowRequest {
+	pub ctx: ActorContext,
+	pub entry_id: Option<String>,
+}
+
 #[derive(Default)]
 pub struct ActorInstanceCallbacks {
 	pub on_migrate: Option<LifecycleCallback<OnMigrateRequest>>,
@@ -298,6 +317,8 @@ pub struct ActorInstanceCallbacks {
 	pub actions: HashMap<String, ActionHandler>,
 	pub on_before_action_response: Option<BeforeActionResponseCallback>,
 	pub run: Option<LifecycleCallback<RunRequest>>,
+	pub get_workflow_history: Option<GetWorkflowHistoryCallback>,
+	pub replay_workflow: Option<ReplayWorkflowCallback>,
 }
 
 impl fmt::Debug for ActorInstanceCallbacks {
@@ -319,6 +340,8 @@ impl fmt::Debug for ActorInstanceCallbacks {
 				&self.on_before_action_response.is_some(),
 			)
 			.field("run", &self.run.is_some())
+			.field("get_workflow_history", &self.get_workflow_history.is_some())
+			.field("replay_workflow", &self.replay_workflow.is_some())
 			.finish()
 	}
 }

rivetkit-rust/packages/rivetkit-core/src/inspector/mod.rs

@@ -1,17 +1,51 @@
+use anyhow::Result;
+use futures::future::BoxFuture;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
 
+type WorkflowHistoryCallback =
+	Arc<dyn Fn() -> BoxFuture<'static, Result<Option<Vec<u8>>>> + Send + Sync>;
+type WorkflowReplayCallback = Arc<
+	dyn Fn(Option<String>) -> BoxFuture<'static, Result<Option<Vec<u8>>>> + Send + Sync,
+>;
+
 #[derive(Clone, Debug, Default)]
 pub struct Inspector(Arc<InspectorInner>);
 
-#[derive(Debug, Default)]
+#[derive(Default)]
 struct InspectorInner {
 	state_revision: AtomicU64,
 	connections_revision: AtomicU64,
 	queue_revision: AtomicU64,
 	active_connections: AtomicU32,
 	queue_size: AtomicU32,
 	connected_clients: AtomicUsize,
+	get_workflow_history: Option<WorkflowHistoryCallback>,
+	replay_workflow: Option<WorkflowReplayCallback>,
+}
+
+impl std::fmt::Debug for InspectorInner {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("InspectorInner")
+			.field("state_revision", &self.state_revision.load(Ordering::SeqCst))
+			.field(
+				"connections_revision",
+				&self.connections_revision.load(Ordering::SeqCst),
+			)
+			.field("queue_revision", &self.queue_revision.load(Ordering::SeqCst))
+			.field(
+				"active_connections",
+				&self.active_connections.load(Ordering::SeqCst),
+			)
+			.field("queue_size", &self.queue_size.load(Ordering::SeqCst))
+			.field(
+				"connected_clients",
+				&self.connected_clients.load(Ordering::SeqCst),
+			)
+			.field("get_workflow_history", &self.get_workflow_history.is_some())
+			.field("replay_workflow", &self.replay_workflow.is_some())
+			.finish()
+	}
 }
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
@@ -29,6 +63,17 @@ impl Inspector {
 		Self::default()
 	}
 
+	pub fn with_workflow_callbacks(
+		get_workflow_history: Option<WorkflowHistoryCallback>,
+		replay_workflow: Option<WorkflowReplayCallback>,
+	) -> Self {
+		Self(Arc::new(InspectorInner {
+			get_workflow_history,
+			replay_workflow,
+			..InspectorInner::default()
+		}))
+	}
+
 	pub fn snapshot(&self) -> InspectorSnapshot {
 		InspectorSnapshot {
 			state_revision: self.0.state_revision.load(Ordering::SeqCst),
@@ -40,6 +85,24 @@ impl Inspector {
 		}
 	}
 
+	pub fn is_workflow_enabled(&self) -> bool {
+		self.0.get_workflow_history.is_some()
+	}
+
+	pub async fn get_workflow_history(&self) -> Result<Option<Vec<u8>>> {
+		let Some(callback) = &self.0.get_workflow_history else {
+			return Ok(None);
+		};
+		callback().await
+	}
+
+	pub async fn replay_workflow(&self, entry_id: Option<String>) -> Result<Option<Vec<u8>>> {
+		let Some(callback) = &self.0.replay_workflow else {
+			return Ok(None);
+		};
+		callback(entry_id).await
+	}
+
 	pub(crate) fn record_state_updated(&self) {
 		self.0.state_revision.fetch_add(1, Ordering::SeqCst);
 	}

rivetkit-rust/packages/rivetkit-core/src/lib.rs

@@ -11,7 +11,8 @@ pub use actor::callbacks::{
 	ActionRequest, ActorInstanceCallbacks, OnBeforeActionResponseRequest,
 	OnBeforeConnectRequest, OnConnectRequest, OnDestroyRequest, OnDisconnectRequest,
 	OnMigrateRequest, OnRequestRequest, OnSleepRequest, OnStateChangeRequest,
-	OnWakeRequest, OnWebSocketRequest, Request, Response, RunRequest,
+	OnWakeRequest, OnWebSocketRequest, ReplayWorkflowRequest, Request, Response,
+	RunRequest, GetWorkflowHistoryRequest,
 };
 pub use actor::config::{
 	ActorConfig, ActorConfigOverrides, CanHibernateWebSocket, FlatActorConfig,

rivetkit-rust/packages/rivetkit-core/src/registry.rs

@@ -31,11 +31,13 @@ use uuid::Uuid;
 
 use crate::actor::action::{ActionDispatchError, ActionInvoker};
 use crate::actor::callbacks::{ActionRequest, OnRequestRequest, OnWebSocketRequest, Request, Response};
+use crate::actor::callbacks::{GetWorkflowHistoryRequest, ReplayWorkflowRequest};
 use crate::actor::config::CanHibernateWebSocket;
 use crate::actor::context::ActorContext;
 use crate::actor::factory::ActorFactory;
 use crate::actor::lifecycle::{ActorLifecycle, StartupOptions};
 use crate::actor::state::{PERSIST_DATA_KEY, PersistedActor, decode_persisted_actor};
+use crate::inspector::Inspector;
 use crate::kv::Kv;
 use crate::sqlite::SqliteDb;
 use crate::types::{ActorKey, ActorKeySegment, SaveStateOpts};
@@ -53,6 +55,7 @@ struct ActiveActorInstance {
 	ctx: ActorContext,
 	factory: Arc<ActorFactory>,
 	callbacks: Arc<crate::actor::callbacks::ActorInstanceCallbacks>,
+	inspector: Inspector,
 }
 
 struct RegistryDispatcher {
@@ -130,6 +133,12 @@ struct InspectorDatabaseExecuteBody {
 	properties: Option<JsonValue>,
 }
 
+#[derive(Debug, Default, Deserialize)]
+#[serde(default, rename_all = "camelCase")]
+struct InspectorWorkflowReplayBody {
+	entry_id: Option<String>,
+}
+
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct InspectorQueueMessageJson {
@@ -262,13 +271,17 @@ impl RegistryDispatcher {
 			.await
 			.map_err(|error| error.into_source())
 			.with_context(|| format!("start actor `{}`", request.actor_id))?;
+		let inspector =
+			build_actor_inspector(request.ctx.clone(), outcome.callbacks.clone());
+		request.ctx.configure_inspector(Some(inspector.clone()));
 
 		let instance = ActiveActorInstance {
 			actor_name: request.actor_name,
 			generation: request.generation,
 			ctx: request.ctx,
 			factory,
 			callbacks: outcome.callbacks,
+			inspector,
 		};
 		let _ = self
 			.active_instances
@@ -482,6 +495,36 @@ impl RegistryDispatcher {
 				};
 				json_http_response(StatusCode::OK, &payload)
 			}
+			(http::Method::GET, "/inspector/workflow-history") => self
+				.inspector_workflow_history(instance)
+				.await
+				.and_then(|(is_workflow_enabled, history)| {
+					json_http_response(
+						StatusCode::OK,
+						&json!({
+							"history": history,
+							"isWorkflowEnabled": is_workflow_enabled,
+						}),
+					)
+				}),
+			(http::Method::POST, "/inspector/workflow/replay") => {
+				let body: InspectorWorkflowReplayBody = match parse_json_body(request) {
+					Ok(body) => body,
+					Err(response) => return Ok(Some(response)),
+				};
+				self
+					.inspector_replay_workflow(instance, body.entry_id)
+					.await
+					.and_then(|(is_workflow_enabled, history)| {
+						json_http_response(
+							StatusCode::OK,
+							&json!({
+								"history": history,
+								"isWorkflowEnabled": is_workflow_enabled,
+							}),
+						)
+					})
+			}
 			(http::Method::GET, "/inspector/traces") => json_http_response(
 				StatusCode::OK,
 				&json!({
@@ -594,18 +637,63 @@ impl RegistryDispatcher {
 			.inspect_messages()
 			.await
 			.context("list queue messages for inspector summary")?;
+		let (is_workflow_enabled, workflow_history) = self
+			.inspector_workflow_history(instance)
+			.await
+			.context("load inspector workflow summary")?;
 		Ok(InspectorSummaryJson {
 			state: decode_cbor_json_or_null(&instance.ctx.state()),
 			is_state_enabled: true,
 			connections: inspector_connections(&instance.ctx),
 			rpcs: inspector_rpcs(instance),
 			queue_size: queue_messages.len().try_into().unwrap_or(u32::MAX),
 			is_database_enabled: instance.ctx.sql().runtime_config().is_ok(),
-			is_workflow_enabled: false,
-			workflow_history: None,
+			is_workflow_enabled,
+			workflow_history,
 		})
 	}
 
+	async fn inspector_workflow_history(
+		&self,
+		instance: &ActiveActorInstance,
+	) -> Result<(bool, Option<JsonValue>)> {
+		let is_workflow_enabled = instance.inspector.is_workflow_enabled();
+		if !is_workflow_enabled {
+			return Ok((false, None));
+		}
+
+		let history = instance
+			.inspector
+			.get_workflow_history()
+			.await
+			.context("load inspector workflow history")?
+			.map(|payload| decode_cbor_json_or_null(&payload))
+			.filter(|value| !value.is_null());
+
+		Ok((true, history))
+	}
+
+	async fn inspector_replay_workflow(
+		&self,
+		instance: &ActiveActorInstance,
+		entry_id: Option<String>,
+	) -> Result<(bool, Option<JsonValue>)> {
+		let is_workflow_enabled = instance.inspector.is_workflow_enabled();
+		if !is_workflow_enabled {
+			return Ok((false, None));
+		}
+
+		let history = instance
+			.inspector
+			.replay_workflow(entry_id)
+			.await
+			.context("replay inspector workflow history")?
+			.map(|payload| decode_cbor_json_or_null(&payload))
+			.filter(|value| !value.is_null());
+
+		Ok((true, history))
+	}
+
 	async fn inspector_database_schema(&self, ctx: &ActorContext) -> Result<JsonValue> {
 		let tables = decode_cbor_json_or_null(
 			&ctx
@@ -1257,6 +1345,57 @@ fn inspector_connections(ctx: &ActorContext) -> Vec<InspectorConnectionJson> {
 		.collect()
 }
 
+fn build_actor_inspector(
+	ctx: ActorContext,
+	callbacks: Arc<crate::actor::callbacks::ActorInstanceCallbacks>,
+) -> Inspector {
+	let get_workflow_history = callbacks.get_workflow_history.as_ref().map(|_| {
+		let callbacks = callbacks.clone();
+		let ctx = ctx.clone();
+		Arc::new(move || -> futures::future::BoxFuture<'static, Result<Option<Vec<u8>>>> {
+			let callbacks = callbacks.clone();
+			let ctx = ctx.clone();
+			Box::pin(async move {
+				let Some(callback) = callbacks.get_workflow_history.as_ref() else {
+					return Ok(None);
+				};
+				callback(GetWorkflowHistoryRequest { ctx }).await
+			})
+		}) as Arc<
+			dyn Fn() -> futures::future::BoxFuture<'static, Result<Option<Vec<u8>>>>
+				+ Send
+				+ Sync,
+		>
+	});
+	let replay_workflow = callbacks.replay_workflow.as_ref().map(|_| {
+		let callbacks = callbacks.clone();
+		let ctx = ctx.clone();
+		Arc::new(
+			move |entry_id: Option<String>| -> futures::future::BoxFuture<
+				'static,
+				Result<Option<Vec<u8>>>,
+			> {
+			let callbacks = callbacks.clone();
+			let ctx = ctx.clone();
+			Box::pin(async move {
+				let Some(callback) = callbacks.replay_workflow.as_ref() else {
+					return Ok(None);
+				};
+				callback(ReplayWorkflowRequest { ctx, entry_id }).await
+			})
+		},
+		) as Arc<
+			dyn Fn(
+					Option<String>,
+				) -> futures::future::BoxFuture<'static, Result<Option<Vec<u8>>>>
+				+ Send
+				+ Sync,
+		>
+	});
+
+	Inspector::with_workflow_callbacks(get_workflow_history, replay_workflow)
+}
+
 fn inspector_rpcs(instance: &ActiveActorInstance) -> Vec<String> {
 	let mut rpcs: Vec<String> = instance.callbacks.actions.keys().cloned().collect();
 	rpcs.sort();