Fast and reliable
background jobs.

func createUser(ctx context.Context, email string) (*User, error) {
	tx, err := dbPool.Begin(ctx)
	if err != nil {
		return nil, err
	}

	defer tx.Rollback(ctx)

	// Create the new user record:
	user, err := insertUser(ctx, tx, email)
	if err != nil {
		return nil, err
	}

	// Enqueue a job to send the user an onboarding email:
	if _, err = riverClient.InsertTx(ctx, tx,
		SendEmailArgs{UserID: user.ID},
		nil,
	); err != nil {
		return nil, err
	}

	// Atomically commit both the user record and the job:
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return user, nil
}

def create_user(email)
  ApplicationRecord.transaction do
    # Create the new user record:
    user = User.create!(email: email)

    # Enqueue on the same Active Record connection:
    river_client.insert River::JobArgsHash.new(
      :send_email, user_id: user.id
    )

    # The user and job commit together when the block returns.
    user
  end
end

async fn create_user(
    pool: &riverqueue::sqlx::PgPool,
    client: &riverqueue::Client,
    email: &str,
) -> Result<User, riverqueue::BoxError> {
    let mut tx = riverqueue::database::begin_postgres(pool).await?;

    // Create the new user record:
    let user = insert_user(&mut tx, email).await?;

    // Enqueue in the same transaction:
    client.insert(SendEmailArgs { user_id: user.id })
        .tx(&mut tx)
        .await?;

    // Atomically commit both the user record and the job:
    tx.commit().await?;
    Ok(user)
}

async function createUser(email: string) {
  const tx = await pool.connect();

  try {
    await tx.query("BEGIN");

    // Create the new user record:
    const user = await insertUser(tx, email);

    // Enqueue in the same transaction:
    await riverClient.insert(sendEmail, { user_id: user.id }, { tx });

    // Atomically commit both the user record and the job:
    await tx.query("COMMIT");
    return user;
  } catch (error) {
    await tx.query("ROLLBACK");
    throw error;
  } finally {
    tx.release();
  }
}