quorlasocial/connector/src/schedule.js

// 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/<slug>/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 { 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";

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); };
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. */
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);
  items.push(post); write(items);
  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);
  write(items); 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;
}

/** 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);
      done.push({ id: p.id, client: p.client, status: p.status, errors: p.errors });
    }
  } finally { fs.rmSync(LOCK, { force: true }); }
  return { published: done };
}