feat(actor): add metadata patching and projection

engine/packages/api-peer/src/actors/delete.rs

@@ -28,6 +28,7 @@ pub async fn delete(ctx: ApiCtx, path: DeletePath, query: DeleteQuery) -> Result
 		ctx.op(pegboard::ops::actor::get::Input {
 			actor_ids: vec![path.actor_id],
 			fetch_error: false,
+			metadata: pegboard::actor_metadata::Projection::None,
 		}),
 		ctx.op(namespace::ops::resolve_for_name_global::Input {
 			name: query.namespace,

engine/packages/api-peer/src/actors/get_or_create.rs

@@ -76,6 +76,7 @@ pub async fn get_or_create(
 					.op(pegboard::ops::actor::get::Input {
 						actor_ids: vec![existing_actor_id],
 						fetch_error: true,
+						metadata: pegboard::actor_metadata::Projection::Full,
 					})
 					.await?;
 

engine/packages/api-peer/src/actors/kv_get.rs

@@ -47,6 +47,7 @@ pub async fn kv_get(ctx: ApiCtx, path: KvGetPath, query: KvGetQuery) -> Result<K
 		.op(pegboard::ops::actor::get::Input {
 			actor_ids: vec![path.actor_id],
 			fetch_error: false,
+			metadata: pegboard::actor_metadata::Projection::None,
 		})
 		.await?;
 

engine/packages/api-peer/src/actors/list.rs

@@ -13,11 +13,12 @@ use rivet_api_types::{actors::list::*, pagination::Pagination};
 )]
 #[tracing::instrument(skip_all)]
 pub async fn list(ctx: ApiCtx, _path: (), query: ListQuery) -> Result<ListResponse> {
-	let key = query.key;
+	let key = query.key.clone();
 	let actor_ids = [
-		query.actor_id,
+		query.actor_id.clone(),
 		query
 			.actor_ids
+			.clone()
 			.map(|x| {
 				x.split(',')
 					.map(|s| {
@@ -35,6 +36,15 @@ pub async fn list(ctx: ApiCtx, _path: (), query: ListQuery) -> Result<ListRespon
 	]
 	.concat();
 	let include_destroyed = query.include_destroyed.unwrap_or(false);
+	let metadata =
+		if !actor_ids.is_empty() || (query.name.is_some() && key.is_some() && !include_destroyed) {
+			pegboard::actor_metadata::Projection::Full
+		} else if query.metadata_key.is_empty() {
+			pegboard::actor_metadata::Projection::None
+		} else {
+			pegboard::actor_metadata::validate_projection_keys(&query.metadata_key)?;
+			pegboard::actor_metadata::Projection::Selected(query.metadata_key.clone())
+		};
 
 	// TODO: Update api-peer to require including the reservation ID in the query if querying with
 	// key in order to assert the request was sent to the correct datacenter
@@ -54,6 +64,7 @@ pub async fn list(ctx: ApiCtx, _path: (), query: ListQuery) -> Result<ListRespon
 			.op(pegboard::ops::actor::get::Input {
 				actor_ids,
 				fetch_error: true,
+				metadata,
 			})
 			.await?;
 
@@ -102,6 +113,7 @@ pub async fn list(ctx: ApiCtx, _path: (), query: ListQuery) -> Result<ListRespon
 					.transpose()?,
 				limit: query.limit.unwrap_or(100),
 				fetch_error: true,
+				metadata,
 			})
 			.await?;
 

engine/packages/api-peer/src/actors/mod.rs

@@ -4,3 +4,4 @@ pub mod get_or_create;
 pub mod kv_get;
 pub mod list;
 pub mod list_names;
+pub mod patch_metadata;

engine/packages/api-peer/src/actors/patch_metadata.rs

@@ -0,0 +1,157 @@
+use anyhow::{Result, bail};
+use gas::prelude::*;
+use rivet_api_builder::ApiCtx;
+use rivet_api_types::actors::patch_metadata::*;
+
+#[utoipa::path(
+	patch,
+	operation_id = "actors_patch_metadata",
+	path = "/actors/{actor_id}/metadata",
+	params(
+		("actor_id" = Id, Path),
+		PatchMetadataQuery,
+	),
+	request_body(content = PatchMetadataRequest, content_type = "application/json"),
+	responses(
+		(status = 200, body = PatchMetadataResponse),
+	),
+)]
+#[tracing::instrument(skip_all)]
+pub async fn patch_metadata(
+	ctx: ApiCtx,
+	path: PatchMetadataPath,
+	query: PatchMetadataQuery,
+	body: PatchMetadataRequest,
+) -> Result<PatchMetadataResponse> {
+	let patch = body
+		.metadata
+		.into_iter()
+		.map(|(key, value)| pegboard::actor_metadata::PatchEntry { key, value })
+		.collect::<Vec<_>>();
+	pegboard::actor_metadata::validate_patch(&patch)?;
+
+	let request_id = Uuid::new_v4().to_string();
+	let mut patched_sub = ctx
+		.subscribe::<pegboard::workflows::actor::MetadataPatched>((
+			"request_id",
+			request_id.clone(),
+		))
+		.await?;
+
+	let (actors_res, namespace_res) = tokio::try_join!(
+		ctx.op(pegboard::ops::actor::get::Input {
+			actor_ids: vec![path.actor_id],
+			fetch_error: false,
+			metadata: pegboard::actor_metadata::Projection::None,
+		}),
+		ctx.op(namespace::ops::resolve_for_name_global::Input {
+			name: query.namespace,
+		}),
+	)?;
+
+	let actor = actors_res
+		.actors
+		.into_iter()
+		.next()
+		.ok_or_else(|| pegboard::errors::Actor::NotFound.build())?;
+	if actor.destroy_ts.is_some() {
+		return Err(pegboard::errors::Actor::NotFound.build());
+	}
+
+	let namespace = namespace_res.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
+	if actor.namespace_id != namespace.namespace_id {
+		return Err(pegboard::errors::Actor::NotFound.build());
+	}
+
+	let res = ctx
+		.signal(pegboard::workflows::actor::PatchMetadata {
+			patch,
+			request_id: Some(request_id),
+		})
+		.to_workflow::<pegboard::workflows::actor::Workflow>()
+		.tag("actor_id", path.actor_id)
+		.graceful_not_found()
+		.send()
+		.await?;
+	if res.is_none() {
+		tracing::warn!(
+			actor_id=?path.actor_id,
+			"actor workflow not found while patching metadata"
+		);
+		return Err(pegboard::errors::Actor::NotFound.build());
+	}
+
+	let patched = patched_sub.next().await?;
+	if let Some(error) = &patched.error {
+		if let Some(actor_error) = rebuild_actor_error(error) {
+			return Err(actor_error);
+		}
+
+		bail!(
+			"actor metadata patch failed: {}:{} {}",
+			error.group,
+			error.code,
+			error.message
+		);
+	}
+
+	Ok(PatchMetadataResponse {})
+}
+
+fn rebuild_actor_error(
+	error: &pegboard::workflows::actor::SerializedError,
+) -> Option<anyhow::Error> {
+	if error.group != "actor" {
+		return None;
+	}
+
+	match error.code.as_str() {
+		"not_found" => Some(pegboard::errors::Actor::NotFound.build()),
+		"metadata_patch_empty" => Some(pegboard::errors::Actor::MetadataPatchEmpty.build()),
+		"metadata_key_invalid" => Some(
+			pegboard::errors::Actor::MetadataKeyInvalid {
+				key_preview: metadata_string_field(error.meta.as_ref(), "key_preview")?,
+			}
+			.build(),
+		),
+		"metadata_key_too_large" => Some(
+			pegboard::errors::Actor::MetadataKeyTooLarge {
+				max_size: metadata_usize_field(error.meta.as_ref(), "max_size")?,
+				key_preview: metadata_string_field(error.meta.as_ref(), "key_preview")?,
+			}
+			.build(),
+		),
+		"metadata_value_too_large" => Some(
+			pegboard::errors::Actor::MetadataValueTooLarge {
+				max_size: metadata_usize_field(error.meta.as_ref(), "max_size")?,
+				key_preview: metadata_string_field(error.meta.as_ref(), "key_preview")?,
+			}
+			.build(),
+		),
+		"metadata_too_large" => Some(
+			pegboard::errors::Actor::MetadataTooLarge {
+				max_size: metadata_usize_field(error.meta.as_ref(), "max_size")?,
+			}
+			.build(),
+		),
+		"metadata_too_many_keys" => Some(
+			pegboard::errors::Actor::MetadataTooManyKeys {
+				max: metadata_usize_field(error.meta.as_ref(), "max")?,
+				count: metadata_usize_field(error.meta.as_ref(), "count")?,
+			}
+			.build(),
+		),
+		_ => None,
+	}
+}
+
+fn metadata_string_field(meta: Option<&serde_json::Value>, field: &str) -> Option<String> {
+	meta?.get(field)?.as_str().map(ToString::to_string)
+}
+
+fn metadata_usize_field(meta: Option<&serde_json::Value>, field: &str) -> Option<usize> {
+	meta?
+		.get(field)?
+		.as_u64()
+		.and_then(|value| usize::try_from(value).ok())
+}

engine/packages/api-peer/src/router.rs

@@ -26,6 +26,10 @@ pub async fn router(
 			.route("/actors", post(actors::create::create))
 			.route("/actors", put(actors::get_or_create::get_or_create))
 			.route("/actors/{actor_id}", delete(actors::delete::delete))
+			.route(
+				"/actors/{actor_id}/metadata",
+				patch(actors::patch_metadata::patch_metadata),
+			)
 			.route("/actors/names", get(actors::list_names::list_names))
 			.route(
 				"/actors/{actor_id}/kv/keys/{key}",

engine/packages/api-public/src/actors/list.rs

@@ -186,6 +186,8 @@ async fn list_inner(ctx: ApiCtx, query: ListQuery) -> Result<ListResponse> {
 			.build());
 		}
 
+		pegboard::actor_metadata::validate_projection_keys(&query.metadata_key)?;
+
 		let limit = query.limit.unwrap_or(100);
 
 		// Fanout to all datacenters

engine/packages/api-public/src/actors/mod.rs

@@ -4,4 +4,5 @@ pub mod get_or_create;
 pub mod kv_get;
 pub mod list;
 pub mod list_names;
+pub mod patch_metadata;
 pub mod utils;

engine/packages/api-public/src/actors/patch_metadata.rs

@@ -0,0 +1,65 @@
+use anyhow::Result;
+use axum::response::{IntoResponse, Response};
+use rivet_api_builder::{
+	ApiError,
+	extract::{Extension, Json, Path, Query},
+};
+use rivet_api_types::actors::patch_metadata::*;
+use rivet_api_util::request_remote_datacenter_raw;
+use rivet_util::Id;
+
+use crate::ctx::ApiCtx;
+
+#[utoipa::path(
+	patch,
+	operation_id = "actors_patch_metadata",
+	path = "/actors/{actor_id}/metadata",
+	params(
+		("actor_id" = Id, Path),
+		PatchMetadataQuery,
+	),
+	request_body(content = PatchMetadataRequest, content_type = "application/json"),
+	responses(
+		(status = 200, body = PatchMetadataResponse),
+	),
+	security(("bearer_auth" = [])),
+)]
+#[tracing::instrument(skip_all)]
+pub async fn patch_metadata(
+	Extension(ctx): Extension<ApiCtx>,
+	Path(path): Path<PatchMetadataPath>,
+	Query(query): Query<PatchMetadataQuery>,
+	Json(body): Json<PatchMetadataRequest>,
+) -> Response {
+	match patch_metadata_inner(ctx, path, query, body).await {
+		Ok(response) => response,
+		Err(err) => ApiError::from(err).into_response(),
+	}
+}
+
+async fn patch_metadata_inner(
+	ctx: ApiCtx,
+	path: PatchMetadataPath,
+	query: PatchMetadataQuery,
+	body: PatchMetadataRequest,
+) -> Result<Response> {
+	ctx.auth().await?;
+
+	if path.actor_id.label() == ctx.config().dc_label() {
+		let res =
+			rivet_api_peer::actors::patch_metadata::patch_metadata(ctx.into(), path, query, body)
+				.await?;
+
+		Ok(Json(res).into_response())
+	} else {
+		request_remote_datacenter_raw(
+			&ctx,
+			path.actor_id.label(),
+			&format!("/actors/{}/metadata", path.actor_id),
+			axum::http::Method::PATCH,
+			Some(&query),
+			Some(&body),
+		)
+		.await
+	}
+}

engine/packages/api-public/src/actors/utils.rs

@@ -83,6 +83,7 @@ pub async fn fetch_actors_by_ids(
 				namespace: namespace.clone(),
 				name: None,
 				key: None,
+				metadata_key: Vec::new(),
 				actor_ids: None,
 				actor_id: dc_actor_ids,
 				include_destroyed,

engine/packages/api-public/src/router.rs

@@ -19,6 +19,7 @@ use crate::{actors, ctx, datacenters, health, metadata, namespaces, runner_confi
 		actors::list_names::list_names,
 		actors::get_or_create::get_or_create,
 		actors::kv_get::kv_get,
+		actors::patch_metadata::patch_metadata,
 		runners::list,
 		runners::list_names,
 		namespaces::list,
@@ -86,6 +87,10 @@ pub async fn router(
 				"/actors/{actor_id}",
 				axum::routing::delete(actors::delete::delete),
 			)
+			.route(
+				"/actors/{actor_id}/metadata",
+				axum::routing::patch(actors::patch_metadata::patch_metadata),
+			)
 			.route(
 				"/actors/names",
 				axum::routing::get(actors::list_names::list_names),