Alert on disconnected senders with a reconnect link

// monitor.mjs
import { readFile, writeFile } from "node:fs/promises";

const API_URL = "https://api.versionseven.ai/v1";
const API_KEY = process.env.VICTORIA_API_KEY;
const SLACK_WEBHOOK_URL = process.env.SLACK_WEBHOOK_URL;
const STATE_FILE = process.env.STATE_FILE ?? "reconnect-state.json";
const PAUSE = process.argv.includes("--pause");
const NO_SENDER_WARNING = "every enabled sender account is disconnected";

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// Sends a request, waiting and retrying on 429 and 503.
async function call(method, path, body) {
  for (let attempt = 1; ; attempt++) {
    const response = await fetch(`${API_URL}${path}`, {
      method,
      headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
      body: body === undefined ? undefined : JSON.stringify(body),
    });
    const data = await response.json().catch(() => ({}));
    if ((response.status !== 429 && response.status !== 503) || attempt === 5) return { status: response.status, data };
    await sleep((Number(response.headers.get("Retry-After")) || 2 ** attempt) * 1000);
  }
}

// Every page of a list endpoint.
async function listAll(path, key) {
  const items = [];
  for (let offset = 0; ; ) {
    const { status, data } = await call("GET", `${path}?limit=500&offset=${offset}`);
    if (status !== 200) throw new Error(`${path} answered ${status} ${data.error}`);
    items.push(...data[key]);
    if (!data.has_more) return items;
    offset += data.count;
  }
}

async function notify(text) {
  if (!SLACK_WEBHOOK_URL) return console.log(text);
  const response = await fetch(SLACK_WEBHOOK_URL, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text }),
    signal: AbortSignal.timeout(10_000),
  });
  if (!response.ok) throw new Error(`Slack answered ${response.status}`);
}

const state = await readFile(STATE_FILE, "utf8").then(JSON.parse).catch(() => ({}));
const now = new Date();
const summary = { disconnected: 0, links_sent: 0, paused: 0 };

const accounts = await listAll("/accounts", "accounts");
for (const account of accounts) {
  const label = account.platform_username ?? account.email ?? account.id;
  if (account.is_active !== false) {
    delete state[account.id]; // reconnected: the next disconnect gets a fresh link
    continue;
  }
  summary.disconnected++;
  const sent = state[account.id];
  if (sent && new Date(sent.expires_at) > now) continue; // already told the team, link still valid

  const { status, data } = await call("POST", "/accounts/connect-link", {
    platform: account.platform === "linkedin" ? "linkedin" : "email",
    display_name: label,
    email: account.email,
    reconnect_account_id: account.id,
  });
  if (status !== 200) {
    console.error(`Couldn't mint a link for ${label}: ${status} ${data.error} ${data.message ?? ""}`);
    continue;
  }
  await notify(
    `:warning: Sender *${label}* (${account.platform}) is disconnected. Reconnect it here before ${data.expires_at.slice(0, 10)}: ${data.url}`
  );
  state[account.id] = { url: data.url, expires_at: data.expires_at, sent_at: now.toISOString() };
  summary.links_sent++;
  await sleep(650);
}

if (PAUSE) {
  const campaigns = (await listAll("/campaigns", "campaigns")).filter((campaign) => campaign.is_active);
  for (const campaign of campaigns) {
    const { status, data } = await call("GET", `/campaigns/${campaign.id}/queue`);
    if (status !== 200 || !data.warnings?.includes(NO_SENDER_WARNING)) continue;
    const paused = await call("PATCH", `/campaigns/${campaign.id}`, { is_active: false });
    if (paused.status !== 200) {
      console.error(`Couldn't pause ${campaign.name}: ${paused.status} ${paused.data.error}`);
      continue;
    }
    await notify(`:pause_button: Paused *${campaign.name}*: every sender on it is disconnected. Activate it again once one is back.`);
    summary.paused++;
    await sleep(650);
  }
}

await writeFile(STATE_FILE, JSON.stringify(state, null, 2));
console.log(JSON.stringify(summary));

# monitor.py
import json
import os
import sys
import time
from datetime import datetime, timezone

import requests

API_URL = "https://api.versionseven.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['VICTORIA_API_KEY']}"}
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")
STATE_FILE = os.environ.get("STATE_FILE", "reconnect-state.json")
PAUSE = "--pause" in sys.argv
NO_SENDER_WARNING = "every enabled sender account is disconnected"


def call(method, path, body=None):
    """Sends a request, waiting and retrying on 429 and 503."""
    for attempt in range(1, 6):
        response = requests.request(method, f"{API_URL}{path}", headers=HEADERS, json=body, timeout=30)
        try:
            data = response.json()
        except ValueError:
            data = {}
        if response.status_code not in (429, 503) or attempt == 5:
            return response.status_code, data
        time.sleep(float(response.headers.get("Retry-After") or 2**attempt))


def list_all(path, key):
    """Every page of a list endpoint."""
    items, offset = [], 0
    while True:
        status, data = call("GET", f"{path}?limit=500&offset={offset}")
        if status != 200:
            raise SystemExit(f"{path} answered {status} {data.get('error')}")
        items.extend(data[key])
        if not data["has_more"]:
            return items
        offset += data["count"]


def notify(text):
    if not SLACK_WEBHOOK_URL:
        print(text)
        return
    response = requests.post(SLACK_WEBHOOK_URL, json={"text": text}, timeout=10)
    response.raise_for_status()


def parse_time(value):
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def main():
    try:
        with open(STATE_FILE, encoding="utf-8") as file:
            state = json.load(file)
    except (FileNotFoundError, ValueError):
        state = {}
    now = datetime.now(timezone.utc)
    summary = {"disconnected": 0, "links_sent": 0, "paused": 0}

    for account in list_all("/accounts", "accounts"):
        label = account.get("platform_username") or account.get("email") or account["id"]
        if account.get("is_active") is not False:
            state.pop(account["id"], None)  # reconnected: the next disconnect gets a fresh link
            continue
        summary["disconnected"] += 1
        sent = state.get(account["id"])
        if sent and parse_time(sent["expires_at"]) > now:
            continue  # already told the team, link still valid

        status, data = call(
            "POST",
            "/accounts/connect-link",
            {
                "platform": "linkedin" if account.get("platform") == "linkedin" else "email",
                "display_name": label,
                "email": account.get("email"),
                "reconnect_account_id": account["id"],
            },
        )
        if status != 200:
            print(f"Couldn't mint a link for {label}: {status} {data.get('error')} {data.get('message', '')}", file=sys.stderr)
            continue
        notify(
            f":warning: Sender *{label}* ({account.get('platform')}) is disconnected. "
            f"Reconnect it here before {data['expires_at'][:10]}: {data['url']}"
        )
        state[account["id"]] = {"url": data["url"], "expires_at": data["expires_at"], "sent_at": now.isoformat()}
        summary["links_sent"] += 1
        time.sleep(0.65)

    if PAUSE:
        for campaign in list_all("/campaigns", "campaigns"):
            if not campaign.get("is_active"):
                continue
            status, data = call("GET", f"/campaigns/{campaign['id']}/queue")
            if status != 200 or NO_SENDER_WARNING not in (data.get("warnings") or []):
                continue
            status, data = call("PATCH", f"/campaigns/{campaign['id']}", {"is_active": False})
            if status != 200:
                print(f"Couldn't pause {campaign['name']}: {status} {data.get('error')}", file=sys.stderr)
                continue
            notify(f":pause_button: Paused *{campaign['name']}*: every sender on it is disconnected. Activate it again once one is back.")
            summary["paused"] += 1
            time.sleep(0.65)

    with open(STATE_FILE, "w", encoding="utf-8") as file:
        json.dump(state, file, indent=2)
    print(json.dumps(summary))


if __name__ == "__main__":
    main()