IFTTT with Alopex DB

What the service does

An IFTTT-like automation service connects apps, devices and services. Users build applets that pair a trigger with an action, such as getting the weather forecast every morning or turning on the security system on leaving home. The data is users, connected services, applet definitions, trigger events and run history.

Where the data lives

SQL tables: users, connected services and applets live in tables. Trigger and action settings differ by service, so they go in a JSON column.
Transactions: incoming trigger events go into a table with a unique constraint on the event identifier. Recording the event and queuing its run commit in the same transaction. However many times the same event arrives, the applet does not run again.
Columnar tables: run records are loaded in batches into a columnar table through HTTP ingest. Success rates per applet and failure trends per service are aggregated in SQL.

The scheduler that checks services on a schedule and the workers that call actions are app code. Both read and write Alopex DB to share state.

Deployment

Alopex DB runs as a server. The endpoint that receives triggers, the scheduler and the workers each connect to the same database over HTTP or gRPC. Operators query the same database in SQL from the CLI. As usage grows, it scales out to an Alopex DB cluster with the same SQL.

Why Alopex DB fits

In an automation service, settings, events and logs, data of very different character, are read and written by many processes at once. Alopex DB handles tables, JSON and columnar logs with the same SQL from HTTP, gRPC or the CLI, and stops duplicates with transactions.

See it in code

A trigger event lands under a unique constraint and commits together with the execution it schedules. Just the core of preventing double runs.

import alopex

db = alopex.Database.open("ifttt.alopex")

db.execute_sql("CREATE TABLE IF NOT EXISTS applets (id INTEGER PRIMARY KEY, name TEXT)")
db.execute_sql("CREATE TABLE IF NOT EXISTS trigger_events (event_id TEXT UNIQUE, applet_id INTEGER)")
db.execute_sql("CREATE TABLE IF NOT EXISTS executions (applet_id INTEGER, state TEXT)")
db.execute_sql("INSERT INTO applets (id, name) VALUES (?, ?)", [1, "雨なら傘を通知"])

# 届いたトリガーを記録し、新しいときだけ実行を予約する（同じイベントで二重実行しない）
def on_trigger(event_id, applet_id):
    tx = db.begin(alopex.TxnMode.READ_WRITE)
    try:
        n = tx.execute_sql(
            "INSERT INTO trigger_events (event_id, applet_id) VALUES (?, ?) ON CONFLICT (event_id) DO NOTHING",
            [event_id, applet_id],
        )
        if n:  # 新しいイベントのときだけ実行を予約する
            tx.execute_sql("INSERT INTO executions (applet_id, state) VALUES (?, ?)", [applet_id, "queued"])
        tx.commit()
    except Exception:
        tx.rollback()
        raise

on_trigger("evt-001", 1)
on_trigger("evt-001", 1)  # 同じイベントが再送されても…
on_trigger("evt-002", 1)

# 予約された実行の数（二重実行していないことを確かめる）
print("実行予約:", db.execute_sql("SELECT COUNT(*) AS n FROM executions")[0]["n"])

However many times the same event arrives, the unique constraint and the transaction make the execution happen exactly once.

An Alopex DB server, with rules in SQL and JSON and run logs in columnar tables, is our pick for an automation service. The quickstart is a good companion.