← back to Norma
agents/instagram-agent/live-media.js
282 lines
/**
* live-media.js — "all the daily Instagram posts" board for norma-instagram (:9810).
*
* MULTI-ACCOUNT: Norma manages 35 IG business accounts, all reachable with ONE shared
* never-expiring META token (accounts.json → token_source META_ACCESS_TOKEN). The board
* has an account dropdown; each account's REAL feed comes straight from the Graph /media
* API, is cached per-account, grouped by day, with delete-any.
*
* Doctrine (matches server.js): a PAGE VIEW never calls the Graph API. A cron + an
* explicit "Refresh" button paginate /media into data/live-media[-<igid>].json; the
* viewer and /api/live/posts only READ that cache.
*
* Security: only accounts present in accounts.json (+ the .env default) can be queried —
* an arbitrary ig_user_id is rejected. Delete reuses posts-api's /api/posts/delete.
*
* Routes (registered before start()'s 404 catch-all):
* GET /live → viewer HTML (public/live-viewer.html)
* GET /api/live/accounts → { accounts:[{handle,ig_user_id,page_name}], default }
* GET /api/live/posts?account=&days= → { account, days:[{date,label,posts}], … }
* POST /api/live/refresh {account,full} → paginate that account newest-first into cache
*/
const fs = require('fs');
const path = require('path');
const https = require('https');
const DIR = __dirname;
const DATA = path.join(DIR, 'data');
const TOMBS = path.join(DATA, 'deleted-posts.jsonl');
const VIEWER = path.join(DIR, 'public', 'live-viewer.html');
const HOST = (process.env.IG_GRAPH_HOST || 'https://graph.facebook.com').replace(/^https?:\/\//, '');
const VER = process.env.IG_GRAPH_VERSION || 'v21.0';
const FIELDS = 'id,caption,media_type,media_url,thumbnail_url,permalink,timestamp,like_count,comments_count';
const TOKEN = () => process.env.IG_ACCESS_TOKEN || ''; // the shared META token
const DEFAULT_ID = () => process.env.IG_USER_ID || '';
// ── account roster (accounts.json: {accounts:[{handle,ig_user_id,page_name,…}]}) ──
function listAccounts() {
let roster = [];
try {
const a = JSON.parse(fs.readFileSync(path.join(DIR, 'accounts.json'), 'utf8'));
const raw = Array.isArray(a.accounts) ? a.accounts : (a.accounts ? Object.values(a.accounts) : []);
roster = raw
.filter((x) => x && x.ig_user_id)
.map((x) => ({ handle: x.handle || x.username || x.ig_user_id, ig_user_id: String(x.ig_user_id), page_name: x.page_name || x.name || '' }));
} catch { /* fall through to env-only */ }
// ensure the .env default (DW) is present + pinned first
const def = DEFAULT_ID();
if (def && !roster.some((r) => r.ig_user_id === def)) {
roster.unshift({ handle: 'designerwallcoverings', ig_user_id: def, page_name: 'Designer Wallcoverings' });
}
// dedupe by id, sort by handle but keep DW first
const seen = new Set();
const uniq = roster.filter((r) => (seen.has(r.ig_user_id) ? false : seen.add(r.ig_user_id)));
uniq.sort((a, b) => (a.ig_user_id === def ? -1 : b.ig_user_id === def ? 1 : a.handle.localeCompare(b.handle)));
return uniq;
}
function resolveAccount(igUserId) {
const roster = listAccounts();
const id = igUserId ? String(igUserId) : DEFAULT_ID();
return roster.find((r) => r.ig_user_id === id) || null; // null = not whitelisted → reject
}
// ── tiny GET-JSON (no deps) — handles BOTH request- and response-stream errors
// (a mid-body ECONNRESET emits 'error' on the response; without a listener it crashes
// the process), with a small retry so transient socket resets self-heal. ──────────────
function getJSONOnce(url) {
return new Promise((resolve, reject) => {
const req = https.get(url, { timeout: 30000 }, (r) => {
let d = '';
r.on('data', (c) => (d += c));
r.on('end', () => { try { resolve(JSON.parse(d)); } catch (e) { reject(e); } });
r.on('error', reject); // ← response-stream error (mid-body reset) — was unhandled
});
req.on('error', reject);
req.on('timeout', () => req.destroy(new Error('request timeout')));
});
}
async function getJSON(url, tries = 3) {
let last;
for (let i = 0; i < tries; i++) {
try { return await getJSONOnce(url); }
catch (e) {
last = e;
if (!/ECONNRESET|timeout|ETIMEDOUT|EAI_AGAIN|socket hang up/i.test(e.message)) break; // only retry transient net errors
await new Promise((r) => setTimeout(r, 800 * (i + 1)));
}
}
throw last;
}
// ── tombstones (shared identity with posts-api: permalink is the stable id) ────
function readTombstones() {
const s = new Set();
if (!fs.existsSync(TOMBS)) return s;
for (const line of fs.readFileSync(TOMBS, 'utf8').trim().split('\n').filter(Boolean)) {
try { const t = JSON.parse(line); if (t.permalink) s.add(t.permalink); if (t.id) s.add(t.id); } catch { /* skip */ }
}
return s;
}
// ── per-account cache (DW default keeps the original live-media.json) ─────────
function cachePathFor(igUserId) {
return igUserId === DEFAULT_ID()
? path.join(DATA, 'live-media.json')
: path.join(DATA, `live-media-${igUserId}.json`);
}
function readCache(igUserId) {
try { return JSON.parse(fs.readFileSync(cachePathFor(igUserId), 'utf8')); }
catch { return { account: null, cached_at: null, complete: false, byId: {} }; }
}
function writeCache(igUserId, c) { fs.writeFileSync(cachePathFor(igUserId), JSON.stringify(c)); }
const REFRESH_LOCKS = new Set(); // per-account refresh lock
/**
* Paginate one account's /media feed newest-first into its cache.
* - default (incremental): stop when a whole page is already-known, or maxPages.
* - { full:true }: walk the entire history (cap 260 pages ≈ 6,500 posts).
*/
async function refreshLiveMedia({ igUserId, full = false, maxPages = full ? 260 : 24 } = {}) {
if (!TOKEN()) throw new Error('IG_ACCESS_TOKEN not set (simulation mode)');
const acct = resolveAccount(igUserId);
if (!acct) throw new Error('unknown account (not in accounts.json)');
const id = acct.ig_user_id;
if (REFRESH_LOCKS.has(id)) throw new Error('a refresh is already running for this account');
REFRESH_LOCKS.add(id);
const token = TOKEN();
const cache = readCache(id);
const known = new Set(Object.keys(cache.byId));
let added = 0, scanned = 0, pages = 0, hitKnown = false;
let url = `https://${HOST}/${VER}/${id}/media?fields=${encodeURIComponent(FIELDS)}&limit=50&access_token=${token}`;
try {
try {
const a = await getJSON(`https://${HOST}/${VER}/${id}?fields=username,name,media_count,followers_count&access_token=${token}`);
if (a && !a.error) cache.account = { ig_user_id: id, handle: acct.handle, username: a.username, name: a.name, media_count: a.media_count, followers_count: a.followers_count };
} catch { /* keep prior */ }
while (url && pages < maxPages) {
const j = await getJSON(url);
if (j.error) throw new Error(`Graph: ${j.error.message}`);
const rows = j.data || [];
pages++;
for (const m of rows) {
scanned++;
if (!cache.byId[m.id]) added++;
cache.byId[m.id] = {
id: m.id, caption: m.caption || '', type: m.media_type || 'IMAGE',
thumb: m.thumbnail_url || m.media_url || '', permalink: m.permalink || '',
ts: m.timestamp || '', likes: m.like_count ?? null, comments: m.comments_count ?? null,
};
}
if (!full && rows.length && rows.every((m) => known.has(m.id))) { hitKnown = true; break; }
url = (j.paging && j.paging.next) || '';
}
cache.cached_at = new Date().toISOString();
if (full && !url) cache.complete = true;
writeCache(id, cache);
return { account: acct.handle, ig_user_id: id, added, scanned, pages, complete: !!cache.complete, hitKnown };
} finally { REFRESH_LOCKS.delete(id); }
}
/**
* Sweep EVERY account in the roster through refreshLiveMedia, serialized (one at a time, with a
* small gap) so we stay gentle on the shared Graph token's rate limit and never trip two refreshes
* of the same account at once. Per-account errors are captured, not thrown — one bad account never
* aborts the sweep. This is what keeps the cross-account calendar LIVE for all accounts, not just DW.
* - incremental (default): each account stops early when it hits already-known posts → cheap.
* - { full:true }: full history walk (used by the backfill safety-net).
* - { onlyIncomplete:true }: skip accounts already marked complete (for the backfill sweep).
* - { max }: cap how many accounts to touch this run (drains a big backfill over several runs).
*/
async function refreshAllAccounts({ full = false, onlyIncomplete = false, max = 0, delayMs = 400 } = {}) {
if (!TOKEN()) return { skipped: 'simulation mode', accounts: 0, added: 0, results: [] };
const roster = listAccounts();
const results = [];
let touched = 0;
for (const a of roster) {
if (onlyIncomplete) {
const c = readCache(a.ig_user_id);
if (c && c.complete === true) continue; // already fully backfilled — skip
}
try {
const r = await refreshLiveMedia({ igUserId: a.ig_user_id, full });
results.push({ handle: a.handle, added: r.added, scanned: r.scanned, complete: r.complete });
} catch (e) {
results.push({ handle: a.handle, error: e.message });
}
touched++;
if (max && touched >= max) break;
await new Promise((r) => setTimeout(r, delayMs));
}
const added = results.reduce((s, r) => s + (r.added || 0), 0);
const errors = results.filter((r) => r.error).length;
return { accounts: results.length, added, errors, results };
}
// ── grouping for the viewer ─────────────────────────────────────────────────
function dayLabel(iso, today) {
const d = new Date(iso);
const key = d.toISOString().slice(0, 10);
const t0 = new Date(today); t0.setHours(0, 0, 0, 0);
const diff = Math.round((t0 - new Date(key + 'T00:00:00')) / 86400000);
let label;
if (diff === 0) label = 'Today';
else if (diff === 1) label = 'Yesterday';
else label = d.toLocaleDateString(undefined, { weekday: 'short', month: 'short', day: 'numeric', year: 'numeric' });
return { key, label };
}
function getLivePosts({ igUserId, days = 0, limit = 0 } = {}) {
const acct = resolveAccount(igUserId);
if (!acct) throw new Error('unknown account (not in accounts.json)');
const cache = readCache(acct.ig_user_id);
const tombs = readTombstones();
const all = Object.values(cache.byId)
.filter((p) => p.permalink && !tombs.has(p.permalink) && !tombs.has(p.id))
.sort((a, b) => (b.ts || '').localeCompare(a.ts || ''));
const cutoff = days > 0 ? Date.now() - days * 86400000 : 0;
let rows = cutoff ? all.filter((p) => new Date(p.ts).getTime() >= cutoff) : all;
if (limit > 0) rows = rows.slice(0, limit);
const now = new Date();
const groups = new Map();
for (const p of rows) {
const { key, label } = dayLabel(p.ts, now);
if (!groups.has(key)) groups.set(key, { date: key, label, posts: [] });
groups.get(key).posts.push(p);
}
return {
account: cache.account || { ig_user_id: acct.ig_user_id, handle: acct.handle, username: acct.handle, name: acct.page_name },
selected: acct.ig_user_id,
cached_at: cache.cached_at,
complete: !!cache.complete,
cached_total: Object.keys(cache.byId).length,
shown: rows.length,
liveEnabled: /^(1|true|yes|on)$/i.test(process.env.IG_LIVE_DELETE || ''),
days: [...groups.values()],
};
}
// ── routes ────────────────────────────────────────────────────────────────────
function registerLiveRoutes(app, agentName = 'instagram-agent') {
app.get('/live', (req, res) => {
if (!fs.existsSync(VIEWER)) return res.status(404).send('live-viewer.html missing');
res.type('html').send(fs.readFileSync(VIEWER, 'utf8'));
});
app.get('/api/live/accounts', (req, res) => {
try {
const accounts = listAccounts().map((a) => {
const c = readCache(a.ig_user_id);
return { ...a, cached_total: Object.keys(c.byId || {}).length, complete: !!c.complete };
});
res.json({ accounts, default: DEFAULT_ID() });
} catch (e) { res.status(500).json({ error: e.message }); }
});
app.get('/api/live/posts', (req, res) => {
try {
res.json(getLivePosts({
igUserId: req.query.account || '',
days: parseInt(req.query.days, 10) || 0,
limit: parseInt(req.query.limit, 10) || 0,
}));
} catch (e) { res.status(/unknown account/.test(e.message) ? 400 : 500).json({ error: e.message }); }
});
app.post('/api/live/refresh', async (req, res) => {
try {
const b = req.body || {};
const full = /^(1|true|yes|on)$/i.test(String(b.full || req.query.full || ''));
const r = await refreshLiveMedia({ igUserId: b.account || req.query.account || '', full });
res.json({ ok: true, ...r, cached_at: new Date().toISOString() });
} catch (e) {
const code = /already running/.test(e.message) ? 409 : /unknown account/.test(e.message) ? 400 : 500;
res.status(code).json({ ok: false, error: e.message });
}
});
}
module.exports = { registerLiveRoutes, refreshLiveMedia, refreshAllAccounts, getLivePosts, listAccounts, readCache, readTombstones };