#!/usr/bin/env python3 """ Standalone flush worker for Hydrogen — drives the queued-work jobs REMOTELY. Unlike bin/notifications-flush.php (which boots the container and therefore has to run ON the production host), this script only needs HTTPS and the admin token: it POSTs to the admin endpoint and lets the server do the work. That is what makes it runnable from any machine. POST {APP_URL}/admin/jobs/flush/{job} Authorization: Bearer ADMIN_API_TOKEN -> 200 { "job": "...", "summary": { ... } } Jobs (the server-side whitelist): - notifications : flushes the notification queue to OneSignal (1 row -> 1 push, N rows -> one digest per recipient). This is the default. - counters : drains the media + user counter buffers. - tracking : drains the click buffer. Config comes from the Hydrogen `.env` (only APP_URL and ADMIN_API_TOKEN are needed — no database, no OneSignal credentials on this machine); every value can be overridden by a real environment variable of the same name. Nothing Hydrogen-specific is imported: this file is fully standalone. Dependencies: pip install requests Usage: python flush_worker.py # flush notifications python flush_worker.py --job counters # another job python flush_worker.py --log /path/w.log # self-log to a file python flush_worker.py --verbose # log even when nothing moved python flush_worker.py --dry-run # resolve config, call nothing Cron — safe at the digest cadence (NOTIFICATION_DIGEST_INTERVAL_MINUTES, 2-5 min). An OS advisory lock makes a run exit immediately if the previous one is still going, and it is released automatically if that run crashes: */2 * * * * /usr/bin/python3 /path/to/bin/flush_worker.py --log /var/log/hydrogen-flush.log Windows Task Scheduler: point it at `pyw` (windowless) with the same arguments. Exit codes: 0 = flushed (or nothing to do), 1 = the flush failed, 2 = bad config or another instance already running. """ from __future__ import annotations import argparse import os import socket import sys import time from datetime import datetime, timezone from pathlib import Path try: import requests except ImportError as exc: # pragma: no cover sys.stderr.write(f"Missing dependency: {exc}. Run: pip install requests\n") sys.exit(2) JOBS = ("notifications", "counters", "tracking") # -------------------------------------------------------------------------- # Config # -------------------------------------------------------------------------- def load_env(env_path: Path) -> dict[str, str]: """Parse a KEY=VALUE .env file (ignores comments / blank lines / `export`).""" values: dict[str, str] = {} if not env_path.is_file(): return values for line in env_path.read_text(encoding="utf-8").splitlines(): line = line.strip() if line == "" or line.startswith("#"): continue if line.startswith("export "): line = line[len("export "):] if "=" not in line: continue key, _, val = line.partition("=") key = key.strip() val = val.strip().strip('"').strip("'") if key: values[key] = val return values class Config: def __init__(self, env: dict[str, str], job: str = "notifications"): def get(key: str, default: str = "") -> str: # A real environment variable always wins over the .env file. return os.environ.get(key, env.get(key, default)) self.api_base = get("APP_URL", "").rstrip("/") self.admin_token = get("ADMIN_API_TOKEN") # A flush walks the whole queue and talks to OneSignal, so it can be far # slower than a regular request — give it room rather than cutting a # push mid-flight. self.timeout = int(get("FLUSH_WORKER_TIMEOUT", "300")) self.log_path = get("FLUSH_WORKER_LOG", "") # Rotate the log file once it reaches this size (bytes); 0 disables it. self.log_max_bytes = int(get("FLUSH_WORKER_LOG_MAX_BYTES", str(5 * 1024 * 1024))) # The lock is PER JOB. Several queues are scheduled side by side (e.g. # notifications every 2 min, counters every 5 min); a single shared lock # would make whichever fires second exit immediately having flushed # nothing, and the collision would be invisible in the log. # An explicit FLUSH_WORKER_LOCK overrides this — keep it per job too, or # the jobs sharing it will starve each other. self.lock_path = get( "FLUSH_WORKER_LOCK", str(Path(os.environ.get("TEMP", "/tmp")) / f"hydrogen-flush-worker-{job}.lock"), ) def require(self) -> None: missing = [ name for name, value in { "APP_URL": self.api_base, "ADMIN_API_TOKEN": self.admin_token, }.items() if value == "" ] if missing: raise SystemExit(f"Missing required config: {', '.join(missing)}") # -------------------------------------------------------------------------- # Single-instance lock — an OS ADVISORY lock (fcntl.flock on POSIX, # msvcrt.locking on Windows). The OS releases it when the holder dies, crash # included, so no stale lock can ever wedge the worker. # -------------------------------------------------------------------------- class InstanceLock: def __init__(self, path: str): self.path = path self.fh = None def __enter__(self) -> "InstanceLock": fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o644) self.fh = os.fdopen(fd, "r+b") try: if os.name == "nt": import msvcrt self.fh.seek(0) msvcrt.locking(self.fh.fileno(), msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(self.fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: self.fh.close() self.fh = None raise SystemExit(f"Another instance is already running (lock {self.path}); exiting.") try: self.fh.seek(0) self.fh.truncate() self.fh.write(str(os.getpid()).encode()) self.fh.flush() except OSError: pass return self def __exit__(self, *_exc) -> None: if self.fh is None: return try: if os.name == "nt": import msvcrt self.fh.seek(0) msvcrt.locking(self.fh.fileno(), msvcrt.LK_UNLCK, 1) else: import fcntl fcntl.flock(self.fh.fileno(), fcntl.LOCK_UN) except OSError: pass self.fh.close() # -------------------------------------------------------------------------- # Logging # -------------------------------------------------------------------------- _LOG_FH = None def log(msg: str, level: str = "INFO") -> None: line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [{level:<5}] {msg}" print(line, flush=True) if _LOG_FH is not None: _LOG_FH.write(line + "\n") _LOG_FH.flush() def _rotate_log(path: Path, max_bytes: int) -> None: """Single-backup size rotation, so a per-minute cron cannot grow the log without bound. No-op when max_bytes <= 0.""" if max_bytes <= 0 or not path.exists(): return try: if path.stat().st_size < max_bytes: return backup = path.with_name(path.name + ".1") if backup.exists(): backup.unlink() path.rename(backup) except OSError: pass # -------------------------------------------------------------------------- # Work # -------------------------------------------------------------------------- def flush(cfg: Config, job: str) -> dict: """POSTs the flush and returns the server's `summary` block.""" resp = requests.post( f"{cfg.api_base}/admin/jobs/flush/{job}", headers={ "Authorization": f"Bearer {cfg.admin_token}", "Accept": "application/json", }, timeout=cfg.timeout, ) if resp.status_code != 200: # Surface the server's own message rather than a bare status code. raise RuntimeError(f"/admin/jobs/flush/{job} -> {resp.status_code}: {resp.text[:300]}") payload = resp.json() summary = payload.get("summary") if not isinstance(summary, dict): raise RuntimeError(f"unexpected response shape: {resp.text[:300]}") return summary def is_idle(summary: dict) -> bool: """True when the flush moved nothing — every leaf counter is 0. Lets a frequent cron stay silent unless something actually happened.""" def leaves(d: dict): for value in d.values(): if isinstance(value, dict): yield from leaves(value) elif isinstance(value, (int, float)): yield value return all(v == 0 for v in leaves(summary)) def describe(summary: dict) -> str: """Flattens the summary into `key=value` pairs (`counters` nests one level). `errors` is left out — it is reported separately, one line each, because the reasons are long sentences.""" parts = [] for key, value in summary.items(): if key == "errors": continue if isinstance(value, dict): inner = " ".join(f"{k}={v}" for k, v in value.items()) parts.append(f"{key}[{inner}]") else: parts.append(f"{key}={value}") return " ".join(parts) def log_errors(summary: dict) -> None: """Prints the per-reason failures the server reported. Without this the operator only sees `failed=N` and has to go dig in the server logs.""" errors = summary.get("errors") if not isinstance(errors, dict): return for reason, count in errors.items(): flat = " ".join(str(reason).split()) log(f" x{count} {flat}", level="ERROR") def report(cfg: Config, job: str, status: str, started_at: float, duration_ms: int, summary: dict | None, message: str | None) -> None: """Reports this tick to `POST /admin/workers/runs`. This worker runs on a different machine from the app, so a crontab that stops firing leaves no trace anywhere the team looks — which is exactly how a wedged flush went unnoticed for days. The heartbeat is what makes the silence visible on `GET /admin/workers`. Best-effort and deliberately silent on failure: telemetry must never turn a successful flush into a failed run, nor add a second error line on top of a real one. A missed heartbeat only costs a gap in the history. `idle` is reported too, not skipped: a tick that found nothing to do is still proof the worker is alive. Reporting only when there is work would make a quiet night indistinguishable from a dead cron. """ try: requests.post( f"{cfg.api_base}/admin/workers/runs", headers={ "Authorization": f"Bearer {cfg.admin_token}", "Accept": "application/json", }, json={ "worker": f"{job}-flush", "status": status, "startedAt": datetime.fromtimestamp(started_at, timezone.utc).isoformat(), "durationMs": duration_ms, "summary": summary, "message": message, "host": socket.gethostname()[:64], }, timeout=10, ) except Exception: pass def main() -> int: parser = argparse.ArgumentParser(description="Hydrogen remote flush worker.") parser.add_argument("--env", default=str(Path(__file__).resolve().parent.parent / ".env"), help="Path to the Hydrogen .env (default: repo .env).") parser.add_argument("--job", choices=JOBS, default="notifications", help="Which queue to flush (default: notifications).") parser.add_argument("--log", default=None, help="Append run output to this file (also printed to stdout). " "Falls back to FLUSH_WORKER_LOG from the env.") parser.add_argument("--verbose", action="store_true", help="Log runs that flushed nothing (silent by default).") parser.add_argument("--dry-run", action="store_true", help="Resolve the config and print the target, without calling it.") args = parser.parse_args() cfg = Config(load_env(Path(args.env)), args.job) cfg.require() log_path = args.log or cfg.log_path if log_path: global _LOG_FH try: p = Path(log_path) p.parent.mkdir(parents=True, exist_ok=True) _rotate_log(p, cfg.log_max_bytes) _LOG_FH = open(log_path, "a", encoding="utf-8") except OSError as exc: print(f"WARNING: cannot open log file {log_path}: {exc}", flush=True) if args.dry_run: log(f"DRY-RUN would POST {cfg.api_base}/admin/jobs/flush/{args.job} (timeout {cfg.timeout}s)") log(f"DRY-RUN lock {cfg.lock_path}") return 0 with InstanceLock(cfg.lock_path): started = time.monotonic() started_at = time.time() try: summary = flush(cfg, args.job) except Exception as exc: elapsed = time.monotonic() - started log(f"{args.job} failed after {elapsed:.1f}s: " f"{type(exc).__name__}: {exc}", level="ERROR") report(cfg, args.job, "failed", started_at, int(elapsed * 1000), None, f"{type(exc).__name__}: {exc}") return 1 elapsed = time.monotonic() - started failed = bool(summary.get("failed")) idle = is_idle(summary) # Reported on EVERY tick, idle included: the heartbeat is the point. # A worker that only reported when it had work would look dead on a # quiet night, which is precisely the signal we want to trust. report(cfg, args.job, "failed" if failed else ("idle" if idle else "ok"), started_at, int(elapsed * 1000), summary, None) # Nothing moved: stay quiet so a 2-minute cron doesn't flood the log. if idle and not args.verbose: return 0 log(f"{args.job} in {elapsed:.1f}s | {describe(summary)}") log_errors(summary) # A tick that pushed nothing because every send failed is not a success. return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())