supervised_worker.py

"""A tiny daemon you can drive from Telegram.

    export TELEGRAM_TOKEN=123:ABC
    export TELEGRAM_CHAT_ID=987654321
    python examples/supervised_worker.py

Then in your chat:  /status   /pause   /resume   /stop   /help
The worker also pushes an alert on every 10th item and a throttled heartbeat.
"""
import asyncio
import os

from telegram_ctl import Controller, Notifier


class Worker:
    def __init__(self):
        self.processed = 0
        self.paused = False
        self.stop = asyncio.Event()


async def work_loop(w: Worker, notify: Notifier):
    while not w.stop.is_set():
        if not w.paused:
            w.processed += 1
            if w.processed % 10 == 0:
                notify.send(f"✅ processed {w.processed} items")
        notify.throttled("heartbeat", f"💓 alive, processed={w.processed}", every=3600)
        await asyncio.sleep(0.5)


async def main():
    if not os.getenv("TELEGRAM_TOKEN"):
        raise SystemExit("set TELEGRAM_TOKEN and TELEGRAM_CHAT_ID first")

    w = Worker()
    notify = Notifier()
    ctl = Controller()

    @ctl.command("status", help="current worker state")
    def status(args):
        state = "paused" if w.paused else "running"
        return f"📊 {state} — processed {w.processed}"

    @ctl.command("pause", help="stop taking new work")
    def pause(args):
        w.paused = True
        return "⏸ paused"

    @ctl.command("resume", help="resume work")
    def resume(args):
        w.paused = False
        return "▶️ resumed"

    @ctl.command("stop", help="graceful shutdown")
    def stop(args):
        w.stop.set()
        return "🛑 stopping"

    notify.send("🚀 worker started — /help for commands")
    await asyncio.gather(work_loop(w, notify), ctl.run(w.stop))


if __name__ == "__main__":
    asyncio.run(main())