// Content queue + scheduler shared by the MCP connector, the dashboard and scripts/run-scheduler.js. // Times are entered in the CLIENT's time zone (clients//client.json → "timezone"), // so a post for a Canadian client goes out at Toronto time even if the team schedules it from India. import fs from "node:fs"; import path from "node:path"; import crypto from "node:crypto"; import { EventEmitter } from "node:events"; import { PROJECT_ROOT } from "./env.js"; import * as meta from "./meta.js"; import * as extra from "./platforms/index.js"; import * as quota from "./platforms/quota.js"; import * as brandkit from "./brandkit.js"; const META_PLATFORMS = ["instagram", "facebook"]; export const ALL_PLATFORMS = [...META_PLATFORMS, ...extra.PLATFORM_IDS]; const DIR = process.env.QUORLA_SCHEDULE_DIR || path.join(PROJECT_ROOT, "data", "schedule"); // override for tests const FILE = path.join(DIR, "queue.json"); const LOCK = path.join(DIR, ".runner.lock"); const LOG = process.env.QUORLA_SCHEDULE_DIR ? path.join(DIR, "publish-log.jsonl") : path.join(PROJECT_ROOT, "data", "publish-log.jsonl"); const CLIENTS = path.join(PROJECT_ROOT, "clients"); fs.mkdirSync(DIR, { recursive: true }); const read = () => { try { return JSON.parse(fs.readFileSync(FILE, "utf8")); } catch { return []; } }; const write = (items) => { const tmp = FILE + ".tmp"; fs.writeFileSync(tmp, JSON.stringify(items, null, 2)); fs.renameSync(tmp, FILE); }; /** Things that happened to posts, for notifications (the dashboard listens; the MCP process doesn't). */ export const events = new EventEmitter(); const emit = (type, post, extra = {}) => { try { events.emit("event", { type, post, ...extra }); } catch { /* listeners must not break publishing */ } }; const log = (entry) => fs.appendFileSync(LOG, JSON.stringify({ at: new Date().toISOString(), ...entry }) + "\n"); const err = (msg, status = 400) => Object.assign(new Error(msg), { status }); // ---------- time zones ---------- function tzOffsetMs(date, tz) { const p = Object.fromEntries(new Intl.DateTimeFormat("en-US", { timeZone: tz, hourCycle: "h23", year: "numeric", month: "2-digit", day: "2-digit", hour: "2-digit", minute: "2-digit", second: "2-digit" }) .formatToParts(date).filter((x) => x.type !== "literal").map((x) => [x.type, Number(x.value)])); return Date.UTC(p.year, p.month - 1, p.day, p.hour, p.minute, p.second) - date.getTime(); } /** "2026-10-20T19:00" in tz → Date (UTC). */ export function zonedToUtc(local, tz) { const m = String(local).match(/^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2})/); if (!m) throw err("Use the format YYYY-MM-DD HH:MM for the time."); const guess = Date.UTC(+m[1], +m[2] - 1, +m[3], +m[4], +m[5]); let t = guess - tzOffsetMs(new Date(guess), tz); t = guess - tzOffsetMs(new Date(t), tz); return new Date(t); } export function formatInTz(date, tz) { return new Intl.DateTimeFormat("en-CA", { timeZone: tz, dateStyle: "medium", timeStyle: "short" }).format(new Date(date)); } // ---------- clients ---------- export function clientInfo(slug) { if (!/^[a-z0-9][a-z0-9-]*$/.test(String(slug || ""))) throw err("Unknown client.", 404); const file = path.join(CLIENTS, slug, "client.json"); let link = {}; try { link = JSON.parse(fs.readFileSync(file, "utf8")); } catch { /* not linked */ } return { slug, name: link.name || slug, pageId: link.pageId || null, igId: link.igId || null, timezone: link.timezone || "America/Toronto", market: link.market || null }; } /** Good default posting slots by market, in the client's local time (adjust per client from insights). */ export function suggestedTimes(tz) { return /Kolkata|Calcutta|Chennai/.test(tz) ? ["12:30", "19:00", "21:00"] // India: lunch, evening, late evening : ["08:00", "12:00", "19:30"]; // North America: commute, lunch, evening } async function accountFor(slug) { const c = clientInfo(slug); if (!c.pageId && !c.igId) throw err(`${c.name} isn't linked to a Facebook Page or Instagram account yet.`); const a = await meta.accountForLink(c); if (!a) throw err(`The agency can't see ${c.name}'s account. Assign it to the social-agent system user.`); return a; } // ---------- queue ---------- const TYPES = ["IMAGE", "REELS", "CAROUSEL", "STORY", "TEXT"]; export function listPosts({ client, status } = {}) { return read().filter((p) => (!client || p.client === client) && (!status || p.status === status)) .sort((a, b) => a.publishAt.localeCompare(b.publishAt)); } export function getPost(id) { const p = read().find((x) => x.id === id); if (!p) throw err("Post not found.", 404); return p; } /** The account a quota is counted against for this client/platform (Page, Instagram or connected account id). */ function quotaAccount(client, platform) { if (platform === "instagram") return clientInfo(client).igId || null; if (platform === "facebook") return clientInfo(client).pageId || null; return extra.vault.connectionFor(client, platform)?.account?.id || null; } /** Warnings (not errors) when a day's planned posts exceed a platform's limit; extra posts then wait for the reset. */ /** Brand-kit and writing checks on every caption a post will use. Naming the agency is refused outright. */ function brandWarnings(post) { const captions = [[null, post.caption], ["facebook", post.fbCaption], ...Object.entries(post.options || {}).map(([pl, o]) => [pl, o?.caption])].filter(([, c]) => c); const out = []; for (const [pl, c] of captions) { for (const w of brandkit.checkCaption(post.client, c, { platforms: pl ? [pl] : post.platforms })) { if (/never name the agency/.test(w.text)) throw err(`${pl ? `${pl} caption: ` : ""}${w.text}`); out.push(`${pl ? `${pl} caption: ` : ""}${w.text}`); } } return [...new Set(out)]; } function quotaWarnings(post, items) { const at = new Date(post.publishAt).getTime(); const out = []; for (const platform of post.platforms) { const lim = quota.limitFor(platform); if (!lim) continue; const acct = quotaAccount(post.client, platform); const w = lim.window === "pt-day" ? quota.windowAt(lim, at) : { start: at - 86400000, end: at + 86400000 }; const planned = items.filter((x) => x.id !== post.id && ["draft", "approved", "publishing"].includes(x.status) && x.platforms.includes(platform) && (lim.scope === "app" || (x.client === post.client)) && new Date(x.publishAt).getTime() >= w.start && new Date(x.publishAt).getTime() < w.end).length; const used = at - Date.now() < 86400000 ? quota.usage(platform, acct)?.used / (lim.cost.publish || 1) || 0 : 0; const msg = quota.plannedWarning(platform, acct, planned + Math.floor(used), at); if (msg) out.push(msg); } return out; } /** * Add a post to the queue as a DRAFT. Nothing is published until a person approves it. * localTime is in the client's time zone, e.g. "2026-10-20 19:00". */ export function createPost({ client, platforms = ["instagram"], type = "IMAGE", media = [], caption = "", fbCaption, collaborators = [], options = {}, localTime, createdBy = "agent", notes } = {}) { const c = clientInfo(client); if (!TYPES.includes(type)) throw err(`Type must be one of ${TYPES.join(", ")}.`); platforms = [...new Set(platforms)].filter((p) => ALL_PLATFORMS.includes(p)); if (!platforms.length) throw err("Pick at least one platform."); // Instagram-only clients (no Facebook Page): drop Facebook when something else is also picked. if (!c.pageId && c.igId && platforms.includes("facebook")) { if (platforms.length === 1) throw err(`${c.name} is Instagram-only (no Facebook Page). Post to Instagram instead.`); platforms = platforms.filter((p) => p !== "facebook"); } if (platforms.some((p) => META_PLATFORMS.includes(p)) && !c.pageId && !c.igId) throw err(`${c.name} isn't linked to a Facebook Page or Instagram account, so it can't post there.`); if (type === "TEXT" && platforms.includes("instagram")) throw err("Instagram needs an image or video. Use Facebook only for text posts."); if (type !== "TEXT" && !media.length) throw err("Add at least one image or video (file path or public URL)."); options = Object.fromEntries(Object.entries(options || {}).filter(([k]) => platforms.includes(k))); for (const p of platforms) if (!META_PLATFORMS.includes(p)) extra.validate(p, c.slug, { type, caption, media, options }); for (const m of media) if (!/^https?:\/\//.test(m) && !fs.existsSync(m)) throw err(`File not found: ${m}`); const publishAt = zonedToUtc(localTime, c.timezone); if (publishAt < Date.now() - 60000) throw err(`That time has already passed in ${c.timezone}.`); const post = { id: crypto.randomUUID().slice(0, 8), client: c.slug, clientName: c.name, platforms, type, media, caption, fbCaption: fbCaption || null, collaborators, options, timezone: c.timezone, localTime: formatInTz(publishAt, c.timezone), publishAt: publishAt.toISOString(), status: "draft", createdBy, createdAt: new Date().toISOString(), notes: notes || null, results: {}, errors: {}, }; const items = read(); post.warnings = [...quotaWarnings(post, items), ...brandWarnings(post)]; items.push(post); write(items); emit("draft_created", post); return post; } export function updatePost(id, patch, by = "team") { const items = read(); const p = items.find((x) => x.id === id); if (!p) throw err("Post not found.", 404); if (!["draft", "failed"].includes(p.status)) throw err("Only drafts or failed posts can be edited. Cancel it first."); for (const k of ["caption", "fbCaption", "media", "platforms", "collaborators", "type", "notes", "options"]) if (patch[k] !== undefined) p[k] = patch[k]; for (const pl of p.platforms) if (!META_PLATFORMS.includes(pl)) extra.validate(pl, p.client, p); if (patch.localTime) { const t = zonedToUtc(patch.localTime, p.timezone); p.publishAt = t.toISOString(); p.localTime = formatInTz(t, p.timezone); } p.status = "draft"; p.updatedBy = by; p.updatedAt = new Date().toISOString(); p.warnings = [...quotaWarnings(p, items), ...brandWarnings(p)]; // Changed after the client looked at it: ask them again. if (p.clientReview && p.clientReview.status !== "pending") p.clientReview = { ...p.clientReview, status: "pending", previous: p.clientReview.status, requestedAt: p.updatedAt, note: "", by: null }; write(items); return p; } /** Put a draft in front of the client (their approval link shows it). The admin's own approval is still needed to publish. */ export function requestClientReview(id, by = "team") { const items = read(); const p = items.find((x) => x.id === id); if (!p) throw err("Post not found.", 404); if (p.status !== "draft") throw err("Only drafts can be sent to the client."); p.clientReview = { status: "pending", requestedBy: by, requestedAt: new Date().toISOString(), note: "", by: null, decidedAt: null }; write(items); emit("client_review_requested", p); return p; } /** The client's answer from their approval link: approved, or changes with a note. */ export function clientDecision(id, client, { decision, note = "", name = "" }) { if (!["approved", "changes"].includes(decision)) throw err("Choose approve or request changes."); if (decision === "changes" && !String(note).trim()) throw err("Tell us what to change."); const items = read(); const p = items.find((x) => x.id === id && x.client === client); if (!p || !p.clientReview) throw err("Post not found.", 404); if (p.status !== "draft") throw err("This post has already moved on; ask your agency if something needs changing."); p.clientReview = { ...p.clientReview, status: decision, note: String(note).slice(0, 2000), by: String(name).slice(0, 80) || "Client", decidedAt: new Date().toISOString() }; log({ action: `client_${decision}`, post: id, client, by: p.clientReview.by }); write(items); emit("client_decision", p); return p; } /** A person approves the post. Facebook parts more than 10 min away are handed to Facebook's own scheduler. */ export async function approvePost(id, by) { if (!by) throw err("Approval needs the name of the person approving."); const items = read(); const p = items.find((x) => x.id === id); if (!p) throw err("Post not found.", 404); if (p.status !== "draft") throw err(`This post is ${p.status}, not a draft.`); p.status = "approved"; p.approvedBy = by; p.approvedAt = new Date().toISOString(); if (p.platforms.includes("facebook") && process.env.ALLOW_PUBLISH === "true" && new Date(p.publishAt) - Date.now() > 15 * 60000) { try { const a = await accountFor(p.client); const r = await meta.fbPublishAny(a, { message: p.fbCaption || p.caption, media: p.media, scheduledAt: new Date(p.publishAt) }); p.results.facebook = { id: r.id || r.post_id, scheduledOnFacebook: true }; log({ action: "fb_scheduled", post: p.id, client: p.client, result: r }); } catch (e) { p.errors.facebook = e.message; } } write(items); return p; } /** An admin retries the platforms that failed. Parts that already posted are skipped, so nothing posts twice. */ export function retryPost(id, by) { if (!by) throw err("Retrying needs the name of the person approving."); const items = read(); const p = items.find((x) => x.id === id); if (!p) throw err("Post not found.", 404); if (!["failed", "partial"].includes(p.status)) throw err(`Only failed posts can be retried; this one is ${p.status}.`); p.status = "approved"; p.retriedBy = by; p.retriedAt = new Date().toISOString(); if (new Date(p.publishAt) < new Date()) p.publishAt = new Date().toISOString(); delete p.retryAt; write(items); return p; } /** Quota use per platform for a client (Connections page): used / limit, and when it resets. */ export function quotaSummary(client) { const c = clientInfo(client); const platforms = [...(c.igId ? ["instagram"] : []), ...(c.pageId ? ["facebook"] : []), ...extra.vault.listConnections({ client }).map((x) => x.platform)]; return Object.fromEntries(platforms.map((pl) => [pl, quota.usage(pl, quotaAccount(client, pl))]).filter(([, u]) => u)); } export async function cancelPost(id, by = "team") { const items = read(); const p = items.find((x) => x.id === id); if (!p) throw err("Post not found.", 404); if (["published", "cancelled"].includes(p.status)) throw err(`This post is already ${p.status}.`); if (p.results.facebook?.scheduledOnFacebook && !p.results.facebook.cancelled) { try { const a = await accountFor(p.client); await meta.graph(p.results.facebook.id, {}, { method: "DELETE", token: a.page_token }); p.results.facebook.cancelled = true; } catch (e) { p.errors.facebook = `Couldn't remove the Facebook scheduled post: ${e.message}. Remove it in Meta Business Suite → Planner.`; } } p.status = "cancelled"; p.cancelledBy = by; p.cancelledAt = new Date().toISOString(); write(items); return p; } /** Meta posts go through the Meta connector; quotas are tracked here the same way as for the other platforms. */ async function publishMeta(platform, a, p) { const acct = platform === "instagram" ? a.instagram?.id : a.page_id; quota.ensure(platform, acct); try { const r = platform === "instagram" ? await meta.igPublishAny(a, { type: p.type, media: p.media, caption: p.caption, collaborators: p.collaborators }) : await meta.fbPublishAny(a, { message: p.fbCaption || p.caption, media: p.media }); quota.record(platform, acct); return r; } catch (e) { throw quota.fromPlatformError(platform, e) || e; } } async function publishNow(p) { let a = null; p.waiting = p.waiting || {}; for (const platform of p.platforms) { if (p.results[platform]?.id && !p.results[platform]?.cancelled) continue; // already done (e.g. FB scheduled natively) try { if (META_PLATFORMS.includes(platform)) a ||= await accountFor(p.client); p.results[platform] = META_PLATFORMS.includes(platform) ? await publishMeta(platform, a, p) : await extra.publish(platform, p.client, p); delete p.errors[platform]; delete p.waiting[platform]; log({ action: "published", platform, post: p.id, client: p.client, result: p.results[platform] }); } catch (e) { p.errors[platform] = e.message; if (e.quota) { p.waiting[platform] = e.resetAt; log({ action: "waiting_for_quota", platform, post: p.id, client: p.client, until: e.resetAt }); } else { delete p.waiting[platform]; log({ action: "failed", platform, post: p.id, client: p.client, error: e.message }); } } } const waits = Object.values(p.waiting); if (waits.length) { // Some platforms are out of quota: keep the post approved and try those again after the earliest reset. p.status = "approved"; p.retryAt = waits.sort()[0]; return; } delete p.retryAt; const ok = p.platforms.filter((x) => p.results[x]?.id).length; p.status = ok === p.platforms.length ? "published" : ok ? "partial" : "failed"; p.publishedAt = new Date().toISOString(); } /** Publish every approved post that is due. Safe to call often; uses a lock so two runners don't double-post. */ export async function runDue({ now = Date.now() } = {}) { if (process.env.ALLOW_PUBLISH !== "true") return { skipped: "ALLOW_PUBLISH is not true in .env" }; try { const st = fs.statSync(LOCK); if (now - st.mtimeMs < 10 * 60000) return { skipped: "another runner is active" }; } catch { /* no lock */ } fs.writeFileSync(LOCK, String(process.pid)); const done = []; try { const items = read(); for (const p of items) { if (p.status !== "approved" || new Date(p.publishAt) > now) continue; if (p.retryAt && new Date(p.retryAt) > now) continue; // waiting for a platform quota to reset p.status = "publishing"; write(items); await publishNow(p); write(items); emit(p.status === "published" ? "published" : p.status === "partial" ? "partly_published" : p.status === "failed" ? "failed" : "waiting", p); done.push({ id: p.id, client: p.client, status: p.status, errors: p.errors }); } } finally { fs.rmSync(LOCK, { force: true }); } return { published: done }; }