LGTM Devlog 28: Game event loop using Google Cloud Scheduler and PubSub

With the quest stage execution loop down, we now need a way to trigger the game event loop regularly to process any new stuff coming in. I'm reviving the "create new game" pubsub trigger function from last time, but updating it with some new stuff we've done recently. The code is in branch 5d28b82

New dependency injection for pubsub events

Just like we add a decorator to decode payloads from HTTP function, I'm adding a decorator to framework.py for PubSub functions:

def inject_pubsub_model(func):
    """ Wrap method with pydantic dependency injection for PubSub functions """

    def wrapper(event: dict, context: Context):
        kwargs = {}
        for arg_name, arg_type in get_type_hints(func).items():
            parse_raw = getattr(arg_type, "parse_raw", None)
            if callable(parse_raw):
                try:
                    kwargs[arg_name] = parse_raw(
                        b64decode(event["data"]).decode("utf-8")
                    )
                except ValidationError as err:
                    raise StatusReturn(error=f"Validation error {err}", http_code=400)

                logger.info(
                    "Decoded model and injected",
                    model=arg_type.__name__,
                    func=func.__name__,
                )

        return func(**kwargs)

    return wrapper

This is very similar to the HTTP one, with two differences

1. we do base64 decoding of the event key first, before putting it to the model
2. there's no return structure handling
3. some differences in the function signature of pubsub trigger events

New Cloud Scheduler task

Over on Google Cloud, we add a task to the Cloud Scheduler service. This scheduler will publish a fixed message to a pubsub topic of our choice on a schedule of our choice.

With the cron of */5 * * * * this schedule will post our desired payload {"source": "Cloud Scheduler"} to the pub/sub topic tick every five minutes. This is our game tick. We can adjust this time later, we might even in the future get rid of it in favour of triggers and hooks, we'll see.

We also need to create a pubsub topic as well.

Pubsub trigger function

The "boilerplate" for our pubsub trigger functions now looks like this:

@inject_pubsub_model
def tick(tick_event: TickEvent):
    """ Game tick """
    logger.info("Tick", source=tick_event.source)

And our GitHub CI gains a new deployment to publish this function

name: Deploy Tick
uses: google-github-actions/deploy-cloud-functions@v0.1.2
with:
  credentials: ${{ secrets.GCP_CREDENTIALS }}
  service_account_email: ${{ secrets.GCP_FUNCTIONS_SERVICE_ACCOUNT }}
  source_dir: '${{ env.workdir }}/app'
  name: tick
  event_trigger_type: providers/cloud.pubsub/eventTypes/topic.publish
  event_trigger_resource: projects/${{ secrets.GCP_PROJECT_ID }}/topics/tick
  runtime: python39
  env_vars: 'APP_VERSION=${{ steps.short_sha.outputs.sha7 }}'

Requisite tests:

@pytest.fixture
def tick_payload():
    """ Payload for tick """
    data = TickEvent(source="test").dict()
    return {
        "context": {
            "eventId": "some-eventId",
            "timestamp": "some-timestamp",
            "eventType": "some-eventType",
            "resource": "some-resource",
        },
        "data": {"data": b64encode(json.dumps(data).encode()).decode()},
    }


def test_bad_payload(tick_client, tick_payload):
    """ Test a bad payload """
    tick_payload["data"]["data"] = b64encode('{"a": 1}'.encode()).decode()
    res = tick_client.post("/", json=tick_payload)
    assert res.status_code != 200


def test_tick(tick_client, tick_payload):
    """ Test tick """
    res = tick_client.post("/", json=tick_payload)
    assert res.status_code == 200

---

With this boilerplate in place, we're ready to implement the game's tick loop, which should be quite simple! Loop through all available quests, and execute their stages.