My partner is really into World of Warcraft. She likes the story, the world, and running around to pick up flowers and crafting materials. There’s actually a really strong economy in the game, and it was the subject of an undergraduate report that I wrote (though I have since moved on to Final Fantasy XIV Online being my MMORPG of choice).

One of the interesting things about that economy is the WoW Token. The idea is that people will purchase a WoW Token for a fixed real-world price (currently $20.00) from an in-game shop. The tokens then get sold on the in-game auction house, at the supply-and-demand price the game determines, and players can redeem them for 30 extra days added to their subscription.

Now, for someone who’s really big into the crafting scene, it’s pretty easy to rack up a lot of gold (the game’s currency). And when the monthly cost of the game is $14.99 a month (or as “low” as $12.99 per month if you buy a year in advance), it’s definitely important to keep appraised of ways to get that time for free.

Which brings us to this project: World of Warcraft recently released a new expansion: Midnight. It adds a ton of new content, ranging from quests to new items. Now, a lot of these items can be found just by exploring the world and completing quests, but the best of them have to be crafted by players who focus on doing that. To craft, you need to gather specific ingredients from across the world. And, as you can imagine, immediately after a new expansion comes out, there’s a big rush to have the best items, both for competitive play and, for some, aesthetics. You can imagine the money that one can make by gathering those items and selling them to people. The market is wide open. And because one can pay for subscription time with that money, it makes that market genuinely valuable.

My partner asked if I could write her a simple little app to send her a notification when the price of the WoW Token dipped below a certain amount… and, me being me, I worked it up into a comprehensive notification engine that’ll let me check any kind of condition and send it to nearly any other platform. Oops.

I wrote a pretty small Python app that runs in a container on my NAS, checks things on a schedule, and pings me on Discord when something interesting happens. I’m calling it the Notification Engine, and it’s designed to be dead simple to extend. You define what to watch (a “stream”), you define where to send the alert (a “provider”), and the Engine handles the rest.

How It Works

The core concept is a simple loop: when the container starts, it loads ./data/streams.yaml, which points to Python files in ./data/streams/. Every X minutes (as defined in the YAML file), it’ll run the check() function in the Python file. If that function says to trigger, it’ll dispatch a notification to the “providers” that have been defined for it.

That’s really it. There’s no database, no web framework, no dependencies beyond requests, PyYAML, and feedparser. It runs in a single process with a while True loop and a time.sleep().

Here’s what a really simple ./data/streams.yaml looks like:

streams:
  wow_token_high:
    providers:
      - discord:
          webhook_url: "https://discord.com/api/webhooks/..."
          webhook_name: "WoW Token Monitor"
          users_to_notify: ['123456789']
    file: "wow_token_high.py"
    check_period: 20
    cooldown_period: 240

streams:
  wow_token_high:
    providers:
      - discord:
          webhook_url: "https://discord.com/api/webhooks/..."
          webhook_name: "WoW Token Monitor"
          users_to_notify: ['123456789']
    file: "wow_token_high.py"
    check_period: 20
    cooldown_period: 240

The check_period is in minutes (so this checks every 20 minutes), and the cooldown_period prevents you from getting spammed if the token stays low, you won’t get pinged again for four hours. Cooldowns are kept in memory, which means a restart resets them, but honestly that’s fine for my use case (especially if one were to be testing that API over and over…).

Writing a Stream

A stream is just a Python file with a check() function that returns a dictionary. At minimum, you need trigger (a boolean) and message (what to send). You can also return an embed dict for Discord rich embeds and a log dict that controls what gets printed to stdout for each state.

Here’s the WoW Token one, stripped down:

def check(name=None, config=None):
    response = requests.get("https://data.wowtoken.app/v2/current/retail.json")
    data = response.json()
    current_price = int(data["us"][1])
    
    return {
        "trigger": current_price < 225000,
        "message": f"WoW Token is under 225k! Currently: **{current_price:,}**",
    }

def check(name=None, config=None):
    response = requests.get("https://data.wowtoken.app/v2/current/retail.json")
    data = response.json()
    current_price = int(data["us"][1])
    
    return {
        "trigger": current_price < 225000,
        "message": f"WoW Token is under 225k! Currently: **{current_price:,}**",
    }

That’s it. If the price is under 225,000 gold, my partner gets a Discord ping in our shared server. If not, the engine moves on and checks again in 20 minutes. And if it stays under 225,000 for the next check, it’ll skip notifying her until that cooldown_period has elapsed. That way, nobody wakes up to 50 notifications in the event of a market crash.

Statefulness

This is where it gets kinda fun: some streams need to remember things between runs. The WoW Token check is stateless – it just looks at the current price – but something like an RSS monitor needs to know what it’s already seen so it doesn’t re-alert you on old items. The impetus for this was getting the most recent Reddit posts from certain subreddits that I follow, and from sites like Game Informer.

I didn’t want to pull in SQLite or Redis for something this small, so I built a tiny key-value store backed by a JSON file. Streams can call get_stream_kv() and set_stream_kv() to persist data under a per-stream namespace. Writes are atomic (write-to-temp-then-rename), which is good enough for a single-process app on a NAS.

In a Stream, you’d do something like:

from state_store import get_stream_kv, set_stream_kv

def check(name=None, config=None):
    last_id = get_stream_kv(name, "last_id", 0)
    # ... do work, find newest_id ...
    set_stream_kv(name, "last_id", newest_id)

from state_store import get_stream_kv, set_stream_kv

def check(name=None, config=None):
    last_id = get_stream_kv(name, "last_id", 0)
    # ... do work, find newest_id ...
    set_stream_kv(name, "last_id", newest_id)

and the resultant state file looks something like:

{
  "streams": {
    "my_rss_feed": {
      "last_seen_epoch": 1709180000
    }
  }
}

It’s not going to win any awards for sophistication, but it works, it’s easy to debug (you can just open the JSON file), and it’s trivially portable.

Providers

Now, as I mentioned earlier, my partner and I have a Discord server together where we chat, share grocery lists and memes, and game together. So, that was the priority. But… as you might have gathered, I don’t like leaving projects half finished. So, I built out a way to easily define other “providers.” Let’s say, for some reason, you’re wanting to send these notifications to Google Chat, Slack, or Microsoft Teams. Just create a file in ./providers/ with a validate() (to define what fields are mandatory for sending the notification) and a send_notification() function, and you’re done. No registration step, no factory pattern, nothing. The engine discovers providers dynamically based on what’s referenced in your YAML.

What’s Next?

There are a few things I want to improve. The timezone is hardcoded to America/New_York right now, which is fine for me but not exactly portable. I’d also like to add a few more providers, maybe Ntfy.sh or plain email, and possibly a way to hot-reload the YAML config without restarting the container.

I plan on putting the code up on GitHub shortly once I’ve had a time to clean it up a bit, but I’m actually quite proud of this one and I was excited to share it, so… hope you’ve enjoyed reading about it!