feat(user): user one-off for users w/o wf

packages/api/auth/src/route/tokens.rs

@@ -190,6 +190,7 @@ async fn fallback_user(
 				::user::workflows::user::Input {
 					user_id,
 					display_name: None,
+					is_already_in_db: false
 				}
 			)
 			.await?

packages/infra/server/Cargo.toml

@@ -76,6 +76,7 @@ pegboard-dc-init.workspace = true
 rivet-cache.workspace = true
 rivet-config.workspace = true
 rivet-connection.workspace = true
+user-workflow-create.workspace = true
 
 [dependencies.sqlx]
 workspace = true

packages/infra/server/src/run_config.rs

@@ -104,6 +104,11 @@ pub fn config(rivet_config: rivet_config::Config) -> Result<RunConfigData> {
 			ServiceKind::Singleton,
 			|config, pools| Box::pin(pegboard_metrics_publish::start(config, pools)),
 		),
+		Service::new(
+			"user_workflow_create",
+			ServiceKind::Oneshot,
+			|config, pools| Box::pin(user_workflow_create::start(config, pools))
+		)
 	];
 
 	if server_config.is_tls_enabled() {

packages/services/cloud/standalone/default-create/src/lib.rs

@@ -42,6 +42,7 @@ pub async fn start(config: rivet_config::Config, pools: rivet_pools::Pools) -> G
 			::user::workflows::user::Input {
 				user_id,
 				display_name: Some(dev_defaults::USER_NAME.into()),
+				is_already_in_db: false
 			}
 		)
 		.tag("user_id", user_id)

packages/services/faker/src/ops/user.rs

@@ -22,6 +22,7 @@ pub async fn user(ctx: &OperationCtx, input: &Input) -> GlobalResult<Output>  {
         user::workflows::user::Input {
             user_id,
             display_name: None,
+            is_already_in_db: false
         }
     )
     .await?

packages/services/user/src/workflows/user/mod.rs

@@ -21,25 +21,28 @@ const UPLOAD_BATCH_SIZE: usize = 256;
 pub struct Input {
 	pub user_id: Uuid,
 	pub display_name: Option<String>,
+	pub is_already_in_db: bool
 }
 
 #[workflow]
 pub async fn user(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResult<()> {
-	let (display_name, _account_number) = ctx.activity(InsertDbInput {
-		user_id: input.user_id,
-		display_name: input.display_name.clone(),
-	}).await?;
+	if !input.is_already_in_db {
+		let (display_name, _account_number) = ctx.activity(InsertDbInput {
+			user_id: input.user_id,
+			display_name: input.display_name.clone(),
+		}).await?;
+
+		ctx.activity(PublishCreationAnalyticsInput {
+			user_id: input.user_id,
+			display_name,
+		}).await?;
+	}
 
 	ctx.msg(CreateComplete {})
 		.tag("user_id", input.user_id)
 		.send()
 		.await?;
 
-	ctx.activity(PublishCreationAnalyticsInput {
-		user_id: input.user_id,
-		display_name,
-	}).await?;
-
 	ctx.repeat(|ctx| {
 		let user_id = input.user_id;
 

packages/services/user/standalone/workflow-create/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "user-workflow-create"
+version.workspace = true
+authors.workspace = true
+license.workspace = true
+edition.workspace = true
+
+[dependencies]
+chirp-client.workspace = true
+chirp-workflow.workspace = true
+futures-util = "0.3"
+rivet-connection.workspace = true
+rivet-config.workspace = true
+rivet-pools.workspace = true
+tokio = { version = "1.40", features = ["full"] }
+uuid = { version = "1", features = ["v4"] }
+user.workspace = true

packages/services/user/standalone/workflow-create/src/lib.rs

@@ -0,0 +1,96 @@
+use chirp_workflow::prelude::*;
+
+const USER_BATCH_SIZE: usize = 128;
+
+#[tracing::instrument(skip_all)]
+pub async fn start(config: rivet_config::Config, pools: rivet_pools::Pools) -> GlobalResult<()> {
+	let client =
+		chirp_client::SharedClient::from_env(pools.clone())?.wrap_new("user-workflow-create");
+	let cache = rivet_cache::CacheInner::from_env(pools.clone())?;
+	let ctx = StandaloneCtx::new(
+		chirp_workflow::compat::db_from_pools(&pools).await?,
+		config,
+		rivet_connection::Connection::new(client, pools, cache),
+		"user-workflow-create",
+	)
+	.await?;
+
+	let (user_count,) = sql_fetch_one!(
+		[ctx, (i64,)]
+		"
+		SELECT COUNT(*)
+		FROM db_users.users
+		WHERE EXISTS(
+			SELECT 1
+			FROM db_user_identity.emails as e
+			WHERE e.user_id = users.user_id
+		)
+		AND NOT EXISTS(
+			SELECT 1
+			FROM db_workflow.workflows
+			WHERE
+				workflow_name = 'user' AND
+				(tags->>'user_id')::UUID = users.user_id
+		)
+		",
+	)
+	.await?;
+
+	if user_count == 0 {
+		return Ok(())
+	}
+
+	for offset in (0..user_count).step_by(USER_BATCH_SIZE) {
+		tracing::debug!(?offset, "creating users");
+
+		let user_ids = sql_fetch_all!(
+			[ctx, (Uuid,)]
+			"
+			SELECT user_id
+			FROM db_user.users
+			WHERE EXISTS(
+				SELECT 1
+				FROM db_user_identity.emails as e
+				WHERE e.user_id = users.user_id
+			)
+			AND NOT EXISTS(
+				SELECT 1
+				FROM db_workflow.workflows
+				WHERE
+					workflow_name = 'user' AND
+					(tags->>'user_id')::UUID = users.user_id
+			)
+			LIMIT $1 OFFSET $2
+			",
+			offset,
+			USER_BATCH_SIZE as i64
+		)
+		.await?
+		.into_iter()
+		.map(|(user_id,)| user_id)
+		.collect::<Vec<_>>();
+
+		if user_ids.len() == 0 {
+			continue;
+		}
+
+		for user_id in user_ids {
+			let mut sub = ctx.subscribe::<
+				user::workflows::user::CreateComplete
+			>(("user_id", user_id)).await?;
+
+			let _ = ctx.workflow(user::workflows::user::Input {
+				user_id,
+				display_name: None,
+				is_already_in_db: true
+			})
+			.tag("user_id", user_id)
+			.dispatch();
+
+			// Await creation completion
+			sub.next().await?;
+		}
+	}
+
+	Ok(())
+}