I Made Tetris in Bluesky

This week, I joined Bluesky, a recently trending social media app. Eventually, I realized it lacked something critical that I once enjoyed on that other micro-blogging site: a sense of community and collaboration. I wanted to build something that could help foster that. This inspired my idea to create a collaborative Tetris game. I call it Bluetris.

What is Bluesky

Bluesky is a social media app that allows users to share images, videos, and short text messages. If you want to follow current events as they unfold, share an interesting fact or observation, or just want to tell a funny story, Bluesky is the place to go. Unlike other social media platforms, Bluesky lets you control your feed directly – you can subscribe to user curated feeds that keep you up to date on whatever topic or personalities you choose. At the time of writing, it has over 22 million users and is growing fast.

While the growth is promising, the real test for the platform is whether or not these users stick around long term. Having prominent users join the platform, like Mark Cuban, is a good sign. However, that may not be enough to grow a healthy community. Rather, it needs something that encourages users to keep coming back. It needs more fun. It needs that sense of excitement that I used to get from the other site. Bluesky needs something that gets people motivated. Maybe I can contribute?

Let’s Build Something

The goal is to create collaborative Tetris. Users will be able to see the board in a post, vote on which move to make next, and then refresh the screen to see result. As I see it, the project can be broken down into three parts:

• Re-create Tetris in python that follows all the rules of logic, but using emoji to render the game state
  • e.g. 🟦🟧🟪🟩🟫🟨🟥⬛⬜
• Utilize the Bluesky API (The AT protocol) to both display the current game state, and also retrieve the top reply for the next move
• Run all this code somewhere in the cloud, and create a scheduler that runs and starts new games

Creating Tetris

I won’t go over everything here, but I will share some of the basics. I started by mocking out what the board should look like.

⬜⬜🟨🟨⬜⬜
⬜⬜⬜🟨🟨⬜ 🟦🟦🟦🟦
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜🟥🟥
🟪⬜⬜🟥🟥🟨
🟪⬜🟧🟧🟨🟨
💥💥💥💥💥💥
🟦🟫🟫🟫🟨⬜
◾◾◾◾◾◾

This was fairly promising. Anyone that has ever played Tetris should be able to understand what’s going on. It also gave me a good starting point to work towards.

Crafting the blocks

First, we need each of the unique pieces that make up the game. This will involve adding the basic rendering and state keeping that our program will need to keep track of.

We can see what each piece, should look like using emoji:

⬜⬜🟪    🟨⬜⬜    ⬜🟫⬜    ⬜🟥🟥    🟨🟨⬜    🟧🟧    🟦🟦🟦🟦
🟪🟪🟪    🟨🟨🟨    🟫🟫🟫    🟥🟥⬜    ⬜🟨🟨    🟧🟧

In python, I represented each piece by creating a new class called Piece. I also decided that I was going to track the type of piece using a variable kind.

class Piece:
	color_list = ["🟦","🟧","🟪","🟩","🟫","🟨","🟥"]
	def __init__(self, kind=0):
		self.kind = kind
		assert kind < len(color_list)
		self.color = color_list[self.kind]
		self.state = []
		self.rotation = 0

To actually represent the state of the piece (which blocks are where), I use a 2d array. I had to create a very long if-else chain to handle all 7 kinds.

		if self.kind == 0: # 4 by 1 (I Block)
			self.state = [[True]*4]
		elif self.kind == 1: # 2 by 2 (O Block)
			self.state = [[True]*2]*2
		elif self.kind == 2: # L shape (L Block)
			self.state = [
				[True, True, False],
				[False, True, False],
				[False, True, False],
			]
		…
		elif self.kind == 5: # Zig-Zag Shape (Z Block)
			self.state = [
				[True, True, False],
				[False, True, True],
				[False]*3,
			]
		...

The only move type the Piece class handles is rotation. I had a lot of trouble getting rotations to work correctly in my first attempts. I could get the 4 by 1 and 2 by 2 blocks to rotate quite easily by simply rotating them. However, the rest of the shapes were more difficult. If I used the same rotation method, then they would appear to rotate about the upper left corner

However, if you represent the pieces as a 3 by 3, padding them with empty ”⬜” blocks, then we got a much better illusion of rotation.

The Board

Next, we need to create another python class to keep track of the game state. This class will also be responsible for the turn structure and physics - such as when the active piece (the falling blocks) collides with something and thus must stop.

class Board:
	def __init__(self, width=6, height=12):
		self.width = width
		self.height = height
		# For each row, map of X-coordinate to Emoji of the block that fell their previously
		self.state = [{} for _ in range(height)]
		# Coordinate of the currently falling piece
		self.active_x = 0
		self.active_y = 0
		self.active_piece = None

Every turn, the active piece must drop one space. Additionally, the board also keeps track of all the blocks as they stack at the bottom. This is essential for filling the board with all the colorful shapes. The board will also handle the other moves that pieces can make. Left and Right translations simply increase and decrease the active_x variable. The board is also responsible for the physics.

	def _check_colision(self, new_x, new_y):
		piece = self.active_piece.to_array()
		for i in range(len(piece)):
			for j in range(len(piece[0])):
				x = new_x + j
				y = new_y + i
				if not piece[i][j]:
					continue
				if (x) in self.state[y]:
					return True
				if x < 0 or x >= self.width:
					return True

If a piece tries to make a move that would either move out of bounds, or would collide with an obstacle, we can revert the move. Thus, we have a working game of Tetris! Now, we just need to post the game to Bluesky!

Bluesky API

In this section, we will build out the basic client for talking to the Bluesky API. We will handle authenticating, posting and retrieving replies, determining the next move, and finally we will handle some issues that I encountered with session timeouts.

The plan for Bluesky is to start a new game every few hours. We will create a new post announcing the game and encouraging users to reply.

NEW GAME

Top liked reply determines
the next move
⬜⬜🟪🟪⬜⬜
⬜⬜⬜🟪⬜⬜ 🟪🟪⬜
⬜⬜⬜🟪⬜⬜ ⬜🟪⬜
⬜⬜⬜⬜⬜⬜ ⬜🟪⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
⬜⬜⬜⬜⬜⬜
◾◾◾◾◾◾

⬅️ ↪️ ⏬ 🔽 ↩️ ➡️
Next turn in 15 minutes!

Users will reply and vote on the next move. After a certain number of minutes, we will retrieving those replies. From the top liked reply, we will extract the next move. Finally, we will update the game state by replying to the last posted game state. We will then repeat the process until the game is over.

Fortunately, Bluesky has an open protocol called the AT Protocol. Because the official client is built in JS, and I was using in python, I needed to find a different package. I initially decided on using Bluesky-Social, however I eventually discovered the much better supported AT Proto python SDK. This will be what I use for the rest of this post.

Setting up the client

from atproto import Client, SessionEvent, Session

bsky_handle = os.environ['BLUESKY_HANDLE']

def init_client():
    client = Client()
    client.on_session_change(on_session_change)

    session_string = get_session()
    if session_string:
        print('Reusing session')
        client.login(session_string=session_string)
    else:
        print('Creating new session')
        bsky_handle = os.environ['BLUESKY_HANDLE']
        client.login(bsky_handle, os.environ['BLUESKY_PASSWORD'])

    return client

Creating the client is fairly easy. I define the handle and password using environmental variables. Bluesky also allows you to create third party app password, which means I can use a different password than I use for the official app.

Creating and replying to posts

Creating a new post is also very easy.

resp = self.client.post(text=game.render())

I use the above code to start every new game. However, what was not so clear to me was how to properly reply to posts. I wanted to create a chain of replies for each game. The idea being that scrolling to the very bottom would allow users to easily follow what is going on. This was not very obvious to me at first glance. The python library I was using did not provide examples for replying. It did show that the post function I was using had a reply_to parameter. After looking through the docs, I figured out that we needed roughly the following structure.

  {
    root: RecordRef,
    parent: RecordRef,
    ...
  }

The way replies seem to work is that every new reply in a thread will keep track of two separate posts. The first is the parent, the post we are directly replying to, and the root, the first post in the chain of replies. For each of these posts, we will need to provide the uri and cid. This ends up looking like the following using our python package:

# Root post
resp = self.client.post(text="Hello World!")
parent_uri = resp.uri
parent_cid = resp.cid
root_uri = resp.uri
root_cid = resp.cid

# Reply post
resp = self.client.post(text="Nice to meet you!", reply_to={
	"root": {
		"uri": root_uri,
		"cid": root_cid,
	},
	"parent": {
		"uri": parent_uri,
		"cid": parent_cid,
	}
})
parent_uri = resp.uri
parent_cid = resp.cid

# Another reply post
resp = self.client.post(text="So much fun!", reply_to={
	"root": {
		"uri": root_uri,
		"cid": root_cid,
	},
	"parent": {
		"uri": parent_uri,
		"cid": parent_cid,
	}
})
parent_uri = resp.uri
parent_cid = resp.cid
# etc etc

I found this slightly clunky, but it was not too hard to track. I would need to make sure that I would always keep the first post’s uri and cid and the same for the last post in a thread. Whenever replying, I would need to update the parent uri and cid based on the response.

Retrieving replies and extracting the next move

Now that I could make replies, I would also need to retrieve replies that users left. The API, app.bsky.feed.getPostThread, is responsible for retrieving all the requests in a post’s reply chain – both above and below. Because I only needed the direct replies, I set depth to 1. I also didn’t need to retrieve the previous posts in the thread, so I set parentHeight to 1 as well.

post = client.get_post_thread(uri=parent_uri, parent_height=1, depth=1)
post_thread = post.thread

From there, I had a fairly easy time extracting the next move. I iterated down all the replies and found the post with the highest like count.

bestMove = "nothing"
maxLikes = -1
for reply in thread.replies:
	post = reply.post
	text = post.record.text
	move = extractMove(text)
	if not move:
		continue
	likeCount = post.like_count
	#repostCount = post.repost_count
	#quoteCount = post.quote_count

	if likeCount > maxLikes:
		maxLikes = likeCount
		bestMove = move
	total_score += score

Further, I also had access to the text of the reply. A simple regex statement allowed me to extract the desired move from the text. I am using a fairly long match pattern:

def extractMove(text):
	pattern = r"⬅️|↪️|⏬|🔽|↩️|➡️|clockwise|cw|counterclockwise|counter[- ]clockwise|ccw|left|right|drop|down|nothing|skip|wait"

	match = re.search(pattern, text or "")
	# etc etc

Refreshing sessions

After I finished implementing the project, I kept running into unauthorized request errors. Eventually, I realized that the access tokens provided by the AT protocol had short lived expiration. When games may last more than an hour, only the first few turns would execute correctly.

Example payload for JSON access token from Bluesky docs

type Payload = {
 iss: // user's DID
 aud: // DID of the service that the request is being made to
 exp: // expiration date of the token, normally set to a short timeframe (<60s)
}

My first thought was to just re-login for each turn. This would reduce the state I would need to store in between turns, and avoid session expiration. Unfortunately, I started seeing an increase in 429 Too Many Requests status codes. I was running into rate limiting. While it might exist elsewhere, I could not find documentation on how logins are rate limited.

Eventually, I found this documentation that let me know how to properly handle sessions. I use a modified version of this code.

def get_session() -> Optional[str]:
    try:
        with open('session.txt') as f:
            return f.read()
    except FileNotFoundError:
        return None


def save_session(session_string: str) -> None:
    with open('session.txt', 'w') as f:
        f.write(session_string)


def on_session_change(event: SessionEvent, session: Session) -> None:
    print('Session changed:', event, repr(session))
    if event in (SessionEvent.CREATE, SessionEvent.REFRESH):
        print('Saving changed session')
        save_session(session.export())

At last, all the integrations with Bluesky were working, and we are ready to deploy the code.

Running Heroku

In this last section, I will go over how I deployed my code on Heroku - at least the major decisions I made. First, I needed a way to start new games every few hours - I ended up creating a service that would handle this for me. Next, I created a task queue using Redis to actually handle running the basic game loop: retrieve replies, make next move, post new game state.

Although Heroku does require paying for services, I was more familiar with it than other platforms. If Bluetris did not take off, I didn’t feel it would be worth learning about a new cloud provider. After all, the main purpose of this project was learning about integrating with Bluesky.

All I would needed from Heroku was a way to start new games every few hours, and a way of queuing tasks for already started games.

Creating a Scheduler

Heroku seems to have changed the way they handle cron jobs. In the past, I remember being able to define custom crontabs that would run at arbitrary intervals. Today, unfortunately, it appears that the options are much more limited. Instead of being able to schedule the job to run every few hours, I instead can specify the job to run either daily, hourly, or every 10 minutes. That is not enough control for me to utilize. After poking around a bit, I eventually found a guide for creating a cron scheduler in a python dyno.

@sched.scheduled_job('cron', day_of_week='mon-fri', hour=17)
def scheduled_job():
    print('This job is run every weekday at 5pm.')

While this looked promising, I wanted to run code every few hours – not at a specific time of the year. It took a while of reading through the docs, but I eventually understood that it uses the regular cron syntax. Now, I could write my own service to execute code every few hours:

# clock.py
from apscheduler.schedulers.blocking import BlockingScheduler

sched = BlockingScheduler()

@sched.scheduled_job('cron', hour="*/4")
def timed_job():
	print('Starting new game…')
	# TODO: start game

sched.start()

Task Queue

Now that I could start new games every few hours, I needed a way to cyclically schedule turns. I didn’t want to rely on using cron jobs because I did not know how long games may last. What if a game takes longer than I expected?

Instead, I will be using a worker based model. Luckily, Heroku provides documentation to do just that! Heroku’s example utilizes queuing via Redis, so I decided to do the same. My implementation would save state on the queue, and then parse it to execute the next turn. Defining the queue was fairly easy.

# worker.py
url = urlparse(os.getenv('REDIS_URL', 'redis://localhost:6379'))
conn = redis.Redis(host=url.hostname, port=url.port, password=url.password, ssl=True, ssl_cert_reqs=None)

queue = Queue('game_queue', connection=conn)

Note: I am using RQ version 1.16.2 – the latest version 2.0.0 changes the interface drastically

To start games, my scheduler script could simply add an empty game state to the queue.

# clock.py
from worker import conn, process_game

@sched.scheduled_job('cron', hour="*/4")
def timed_job():
	print('Starting new game...')
	q = Queue('game_queue', connection=conn)
	# process_game helper function to be defined later
	job = q.enqueue(process_game, None)
	print("Started job:", job.key)

Next, I created a worker that would listen to the task queue:

# worker.py
from RQ import Worker, Queue, Connection

if __name__ == '__main__':
    # Start the worker
    with Connection(conn):
        worker = Worker([queue])
        worker.work(with_scheduler=True) # with_scheduler is required for adding delays to events in the queue

Initially, I was very confused by the RQ worker. While creating the connection was straight forward, I did not understand how it determined which function would handle the event. The key to understanding RQ is that you must provide a function pointer to the handler when you create the event.

# Define the task to process the game state
def process_game(bsky_state):
    print("Pulled from queue...")
    g = GameController.import(bsky_state)

I built the game controller to handle serializing and de-serializing the game state. This allows me to keep all data required for running the game within the event queue. When the provided state is None, that means that it is the start of the new game - the controller will start a new thread.

Next, all we need to do step the game controller. This will call all the logic I’ve previously gone over. This includes: retrieving the replies, running all the physics, rendering the game, and then posting the result as a reply in the thread.

    isFinished, delay = g.stepGame()
    g.postReply()

Finally, we need to queue the next turn. Because we want to give a few minutes between turns, the game controller will let us know what delay we should add to the queue. With RQ, this is easy to implement!

	if not b.finished:
		new_state = g.export()
		job = queue.enqueue_in(delay, process_game, new_state)
		print("Added next game turn to queue:", job.key)
	else:
		print("Game is finished!")

Finally, we’ve put all the pieces together! Our bot will run Tetris in Bluesky! I should also mention that I ran into issues with the Redis instance timing out. In order to handle this case, I included retry logic in the cron scheduler to wake up Redis as needed.

Conclusion

I’m excited by the potential for fun and collaborative games on Bluesky. Even though Bluetris might not go viral, the experience has expanded my coding skills and opened up new possibilities for integrating social media with gaming. If you are interested, you can check out the final product below:

Bluesky Plays Tetris

There are a few bugs/issues I may need to fix, but it should be stable enough to use. I have some ideas for quality of life improvements, but I will wait to see what the interest is before I pursue them.

If you want, you can also follow me on Bluesky. Lastly, I plan on publishing either the entire or some subset of the code base. It’s in a very messy state at the moment, so I will need to clean it up before I am comfortable sharing it. I will update the post to include it when I am ready.

I do not yet know if Bluetris will have an impact on the platform. It may never encourage the long term engagement and community square atmosphere that I believe Bluesky needs to succeed. Regardless, I hope others might take inspiration from my work and build on it to create even more engaging experiences. Please let me know if you do so! I am very excited to see what people create.