Keep campaigns fed from your own lead list

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

const API_URL = "https://api.versionseven.ai/v1";
const API_KEY = process.env.VICTORIA_API_KEY;
const MIN_DAYS = Number(process.env.MIN_DAYS ?? 7);
const STATE_FILE = process.env.STATE_FILE ?? "refill-state.json";
const [listFile = "leads.json"] = process.argv.slice(2);

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

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

// Your lead source. Replace the body of this function with a query, a sheet read or an export.
async function leadsFor(campaignId) {
  const lists = JSON.parse(await readFile(listFile, "utf8"));
  return lists[campaignId] ?? [];
}

const leadKey = (lead) => (lead.email ?? lead.linkedin_url ?? "").trim().toLowerCase().replace(/\/+$/, "");

const state = await readFile(STATE_FILE, "utf8").then(JSON.parse).catch(() => ({}));
const summary = { checked: 0, refilled: 0, enrolled: 0, skipped: 0 };

const { data: list } = await call("GET", "/campaigns?limit=500");
for (const campaign of list.campaigns.filter((candidate) => candidate.is_active)) {
  summary.checked++;
  const { status, data: queue } = await call("GET", `/campaigns/${campaign.id}/queue`);
  if (status !== 200) {
    console.error(`${campaign.name}: queue answered ${status} ${queue.error}`);
    continue;
  }
  if (queue.days_of_runway === null) {
    console.log(`${campaign.name}: can't send (${queue.warnings.join("; ")}); adding leads wouldn't help`);
    continue;
  }
  if (queue.days_of_runway >= MIN_DAYS) {
    console.log(`${campaign.name}: ${queue.days_of_runway} days of runway, nothing to do`);
    continue;
  }

  const needed = Math.ceil((MIN_DAYS - queue.days_of_runway) * queue.daily_capacity);
  const sent = new Set(state[campaign.id] ?? []);
  const batch = (await leadsFor(campaign.id)).filter((lead) => leadKey(lead) && !sent.has(leadKey(lead))).slice(0, needed);
  console.log(`${campaign.name}: ${queue.days_of_runway} days of runway, wants ${needed} leads, list has ${batch.length} new`);
  summary.refilled++;

  for (const lead of batch) {
    const body = { campaign_id: campaign.id, lead };
    const idempotencyKey = createHash("sha256").update(`${campaign.id}:${leadKey(lead)}`).digest("hex");
    const { status, data } = await call("POST", "/leads", body, { "Idempotency-Key": idempotencyKey });
    sent.add(leadKey(lead)); // recorded whatever the answer, so the lead isn't retried every day
    if (status === 201 || status === 200) summary.enrolled++;
    else {
      summary.skipped++;
      console.log(`  ${leadKey(lead)}: ${status} ${data.error}`);
    }
    await sleep(650);
  }
  state[campaign.id] = [...sent];
}

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

# refill.py
import hashlib
import json
import os
import sys
import time

import requests

API_URL = "https://api.versionseven.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['VICTORIA_API_KEY']}"}
MIN_DAYS = float(os.environ.get("MIN_DAYS", 7))
STATE_FILE = os.environ.get("STATE_FILE", "refill-state.json")
LIST_FILE = sys.argv[1] if len(sys.argv) > 1 else "leads.json"


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


def leads_for(campaign_id):
    """Your lead source. Replace this with a query, a sheet read or an export."""
    with open(LIST_FILE, encoding="utf-8") as file:
        return json.load(file).get(campaign_id, [])


def lead_key(lead):
    return (lead.get("email") or lead.get("linkedin_url") or "").strip().lower().rstrip("/")


def main():
    try:
        with open(STATE_FILE, encoding="utf-8") as file:
            state = json.load(file)
    except (FileNotFoundError, ValueError):
        state = {}
    summary = {"checked": 0, "refilled": 0, "enrolled": 0, "skipped": 0}

    _, listing = call("GET", "/campaigns?limit=500")
    for campaign in [c for c in listing["campaigns"] if c.get("is_active")]:
        summary["checked"] += 1
        status, queue = call("GET", f"/campaigns/{campaign['id']}/queue")
        if status != 200:
            print(f"{campaign['name']}: queue answered {status} {queue.get('error')}", file=sys.stderr)
            continue
        if queue["days_of_runway"] is None:
            print(f"{campaign['name']}: can't send ({'; '.join(queue['warnings'])}); adding leads wouldn't help")
            continue
        if queue["days_of_runway"] >= MIN_DAYS:
            print(f"{campaign['name']}: {queue['days_of_runway']} days of runway, nothing to do")
            continue

        needed = int(-(-(MIN_DAYS - queue["days_of_runway"]) * queue["daily_capacity"] // 1))  # ceil
        sent = set(state.get(campaign["id"], []))
        batch = [lead for lead in leads_for(campaign["id"]) if lead_key(lead) and lead_key(lead) not in sent][:needed]
        print(f"{campaign['name']}: {queue['days_of_runway']} days of runway, wants {needed} leads, list has {len(batch)} new")
        summary["refilled"] += 1

        for lead in batch:
            key = hashlib.sha256(f"{campaign['id']}:{lead_key(lead)}".encode()).hexdigest()
            status, data = call("POST", "/leads", {"campaign_id": campaign["id"], "lead": lead}, {"Idempotency-Key": key})
            sent.add(lead_key(lead))  # recorded whatever the answer, so the lead isn't retried every day
            if status in (200, 201):
                summary["enrolled"] += 1
            else:
                summary["skipped"] += 1
                print(f"  {lead_key(lead)}: {status} {data.get('error')}")
            time.sleep(0.65)
        state[campaign["id"]] = sorted(sent)

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


if __name__ == "__main__":
    main()