discgod/extension/background.js

/*
 * discgod — background service worker (the engine).
 *
 * Holds the Discogs API token, runs the schedule, paginates each watched
 * seller's full inventory from api.discogs.com (rate-limited, client-side so it
 * uses *your* token's 60/min quota), and POSTs the raw listing pages to the
 * ingest server, which maps + upserts them into discogs_full and detects price
 * changes / sales.
 *
 * The watchlist itself lives server-side (discgod_seller_watch) so the popup,
 * this worker, and the headless cron crawler all agree on who to track.
 */

const DEFAULTS = {
  token: "",
  serverUrl: "http://100.91.239.7:5057",
  intervalHours: 12,
  status: "For Sale",
  perPage: 100,
  baseDelayMs: 1100,        // ~55 Discogs calls/min, under the 60/min ceiling
  autoCrawl: true,
};

const ALARM_CRAWL = "discgod-crawl";     // periodic full sweep
const ALARM_KEEPALIVE = "discgod-tick";  // resurrects the worker mid-crawl / drains queue
const QUEUE_KEY = "sendQueue";           // failed server POSTs, retried later

// ---------------------------------------------------------------------------
// Config
// ---------------------------------------------------------------------------
async function getConfig() {
  const stored = await chrome.storage.local.get(Object.keys(DEFAULTS));
  return { ...DEFAULTS, ...stored };
}
async function setConfig(patch) {
  await chrome.storage.local.set(patch);
  if ("intervalHours" in patch) await scheduleCrawl();
  return getConfig();
}

// ---------------------------------------------------------------------------
// Lifecycle
// ---------------------------------------------------------------------------
chrome.runtime.onInstalled.addListener(scheduleCrawl);
chrome.runtime.onStartup.addListener(scheduleCrawl);

async function scheduleCrawl() {
  const { intervalHours, autoCrawl } = await getConfig();
  await chrome.alarms.clear(ALARM_CRAWL);
  if (autoCrawl) {
    const period = Math.max(15, Math.round(intervalHours * 60)); // minutes; min 15
    chrome.alarms.create(ALARM_CRAWL, { periodInMinutes: period, delayInMinutes: 1 });
  }
}

chrome.alarms.onAlarm.addListener(async (alarm) => {
  if (alarm.name === ALARM_CRAWL) {
    crawlAll().catch((e) => log("crawlAll failed", e.message));
  } else if (alarm.name === ALARM_KEEPALIVE) {
    // Worker was revived (or never died) mid-crawl: keep draining.
    drainQueue();
    const { crawling } = await chrome.storage.local.get("crawling");
    if (crawling) processQueue().catch(() => {});
  }
});

function keepaliveOn() {
  chrome.alarms.create(ALARM_KEEPALIVE, { periodInMinutes: 1 });
}
function keepaliveOff() {
  chrome.alarms.clear(ALARM_KEEPALIVE);
}

// ---------------------------------------------------------------------------
// Discogs API (client-side, authenticated, rate-limited)
// ---------------------------------------------------------------------------
async function discogsGet(path, params) {
  const cfg = await getConfig();
  if (!cfg.token) throw new Error("No Discogs token set");
  const url = new URL("https://api.discogs.com" + path);
  if (params) for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);

  for (let attempt = 0; attempt < 5; attempt++) {
    const resp = await fetch(url, {
      headers: { Authorization: `Discogs token=${cfg.token}`, Accept: "application/json" },
    });
    if (resp.status === 429) {
      const wait = Math.min(60000, 2000 * 2 ** attempt);
      await sleep(wait);
      continue;
    }
    if (resp.status === 401) throw new Error("401 — bad or missing token");
    if (resp.status === 404) throw new Error("404 — no such user/inventory");
    if (!resp.ok) {
      await sleep(1500 * (attempt + 1));
      continue;
    }
    const remaining = parseInt(resp.headers.get("X-Discogs-Ratelimit-Remaining") || "60", 10);
    await sleep(remaining <= 2 ? 5000 : cfg.baseDelayMs);
    return resp.json();
  }
  throw new Error(`Discogs API gave up: ${path}`);
}

// Quick token check — returns the authenticated identity.
async function testToken() {
  const me = await discogsGet("/oauth/identity");
  return { ok: true, username: me.username };
}

// ---------------------------------------------------------------------------
// Server I/O (the thing that writes Postgres). Failed posts queue + retry.
// ---------------------------------------------------------------------------
async function serverFetch(endpoint, { method = "GET", body } = {}) {
  const { serverUrl } = await getConfig();
  const resp = await fetch(serverUrl + endpoint, {
    method,
    headers: { "Content-Type": "application/json" },
    body: body ? JSON.stringify(body) : undefined,
  });
  if (!resp.ok) throw new Error(`server ${endpoint} -> ${resp.status}`);
  return resp.json();
}

// Post to the ingest server; on failure, queue for later retry so a momentary
// server/network blip never loses a captured page.
async function postOrQueue(endpoint, body) {
  try {
    return await serverFetch(endpoint, { method: "POST", body });
  } catch (e) {
    const { [QUEUE_KEY]: q = [] } = await chrome.storage.local.get(QUEUE_KEY);
    q.push({ endpoint, body, attempts: 0, ts: Date.now() });
    await chrome.storage.local.set({ [QUEUE_KEY]: q });
    setBadge(String(q.length), "#c0392b");
    throw e;
  }
}

async function drainQueue() {
  const { [QUEUE_KEY]: q = [] } = await chrome.storage.local.get(QUEUE_KEY);
  if (!q.length) return;
  const keep = [];
  for (const item of q) {
    try {
      await serverFetch(item.endpoint, { method: "POST", body: item.body });
    } catch {
      item.attempts++;
      if (item.attempts < 10) keep.push(item);
    }
  }
  await chrome.storage.local.set({ [QUEUE_KEY]: keep });
  setBadge(keep.length ? String(keep.length) : "", keep.length ? "#c0392b" : "#2e7d32");
}

// ---------------------------------------------------------------------------
// Crawl one seller: paginate their whole inventory, post each page, finish.
// ---------------------------------------------------------------------------
async function crawlSeller(username, filter) {
  const cfg = await getConfig();
  filter = filter || {};
  const crawlId = `${username}-${Date.now()}`;

  // Profile gives seller_id + a reliable num_for_sale for the snapshot.
  let profile = null, sellerId = null;
  try {
    profile = await discogsGet(`/users/${encodeURIComponent(username)}`);
    sellerId = profile.id;
  } catch (e) {
    log(`profile ${username}`, e.message);
  }

  let page = 1, pages = 1, total = null;
  const baseParams = {
    per_page: cfg.perPage, status: cfg.status, sort: "listed", sort_order: "desc", ...filter,
  };

  while (page <= pages) {
    const data = await discogsGet(`/users/${encodeURIComponent(username)}/inventory`,
      { ...baseParams, page });
    const pg = data.pagination || {};
    pages = pg.pages || 1;
    total = pg.items ?? total;

    await postOrQueue("/discgod/ingest", {
      username, seller_id: sellerId,
      seller: page === 1 ? profile : null,
      page, total, crawl_id: crawlId, filter,
      listings: data.listings || [],
    });
    setBadge("⟳", "#1565c0");
    page++;
  }

  const filtered = Object.keys(filter).length > 0;
  await postOrQueue("/discgod/crawl_complete", {
    username, crawl_id: crawlId, filter,
    treat_complete: !filtered, listing_count: total,
  });
  return total;
}

// ---------------------------------------------------------------------------
// Crawl every watched seller (queue-driven so it survives worker restarts).
// ---------------------------------------------------------------------------
async function crawlAll() {
  let tracked;
  try {
    const res = await serverFetch("/discgod/tracked");
    tracked = (res.sellers || []).filter((s) => s.enabled);
  } catch (e) {
    log("cannot load watchlist", e.message);
    return;
  }
  const queue = tracked.map((s) => ({ username: s.username, filter: s.filter_params || {} }));
  await chrome.storage.local.set({ crawlQueue: queue, crawling: true });
  await processQueue();
}

async function crawlOne(username) {
  await chrome.storage.local.set({
    crawlQueue: [{ username, filter: {} }], crawling: true,
  });
  await processQueue();
}

// Drains crawlQueue one seller at a time, persisting progress between sellers.
let processing = false;
async function processQueue() {
  if (processing) return;
  processing = true;
  keepaliveOn();
  try {
    for (;;) {
      const { crawlQueue = [] } = await chrome.storage.local.get("crawlQueue");
      if (!crawlQueue.length) break;
      const job = crawlQueue[0];
      try {
        const n = await crawlSeller(job.username, job.filter);
        log(`crawled ${job.username}`, `${n} listings`);
      } catch (e) {
        log(`crawl ${job.username} failed`, e.message);
      }
      const { crawlQueue: cur = [] } = await chrome.storage.local.get("crawlQueue");
      cur.shift();
      await chrome.storage.local.set({ crawlQueue: cur, lastCrawlAt: Date.now() });
    }
    await drainQueue();
  } finally {
    await chrome.storage.local.set({ crawling: false });
    keepaliveOff();
    setBadge("", "#2e7d32");
    processing = false;
    notify("discgod", "Inventory sweep complete");
  }
}

// ---------------------------------------------------------------------------
// Messages from the popup / content script
// ---------------------------------------------------------------------------
chrome.runtime.onMessage.addListener((msg, _sender, sendResponse) => {
  (async () => {
    try {
      switch (msg.type) {
        case "getConfig":   return sendResponse(await getConfig());
        case "setConfig":   return sendResponse(await setConfig(msg.patch));
        case "testToken":   return sendResponse(await testToken());
        case "getStats":    return sendResponse(await serverFetch("/discgod/stats"));
        case "getTracked":  return sendResponse(await serverFetch("/discgod/tracked"));
        case "track":       return sendResponse(await serverFetch("/discgod/track",
                              { method: "POST", body: msg.body }));
        case "untrack":     return sendResponse(await serverFetch("/discgod/untrack",
                              { method: "POST", body: { username: msg.username } }));
        case "scrapeAll":   crawlAll(); return sendResponse({ ok: true, started: true });
        case "scrapeOne":   crawlOne(msg.username); return sendResponse({ ok: true, started: true });
        case "status":      return sendResponse(await statusSnapshot());
        default:            return sendResponse({ ok: false, error: "unknown message" });
      }
    } catch (e) {
      sendResponse({ ok: false, error: e.message });
    }
  })();
  return true; // async
});

async function statusSnapshot() {
  const { crawling, crawlQueue = [], lastCrawlAt, [QUEUE_KEY]: q = [] } =
    await chrome.storage.local.get(["crawling", "crawlQueue", "lastCrawlAt", QUEUE_KEY]);
  return { ok: true, crawling: !!crawling, remaining: crawlQueue.length,
           queued: q.length, lastCrawlAt: lastCrawlAt || null };
}

// ---------------------------------------------------------------------------
// Small utilities
// ---------------------------------------------------------------------------
function sleep(ms) { return new Promise((r) => setTimeout(r, ms)); }

function setBadge(text, color) {
  chrome.action.setBadgeText({ text: text || "" });
  if (color) chrome.action.setBadgeBackgroundColor({ color });
}

function notify(title, message) {
  try {
    chrome.notifications.create({
      type: "basic", iconUrl: "icons/icon128.png", title, message,
    });
  } catch { /* notifications optional */ }
}

function log(label, detail) {
  console.log(`[discgod] ${label}${detail ? ": " + detail : ""}`);
}