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";

const DIR = path.join(PROJECT_ROOT, "data", "schedule");
const FILE = path.join(DIR, "queue.json");
const LOCK = path.join(DIR, ".runner.lock");
const LOG = 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, 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) throw err(`${c.name} isn't linked to a Facebook Page yet.`);
  const a = (await meta.listAccounts()).find((x) => x.page_id === c.pageId) || (await meta.listAccounts({ refresh: true })).find((x) => x.page_id === c.pageId);
  if (!a) throw err(`The agency can't see ${c.name}'s Page. 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;
}

/**
 * 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 = [], 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) => p === "instagram" || p === "facebook");
  if (!platforms.length) throw err("Pick Instagram, Facebook or both.");
  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).");
  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, 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(); 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"]) if (patch[k] !== undefined) p[k] = patch[k];
  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();
  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;
}

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;
}

async function publishNow(p) {
  const a = await accountFor(p.client);
  for (const platform of p.platforms) {
    if (p.results[platform]?.id && !p.results[platform]?.cancelled) continue; // already done (e.g. FB scheduled natively)
    try {
      p.results[platform] = 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 });
      delete p.errors[platform];
      log({ action: "published", platform, post: p.id, client: p.client, result: p.results[platform] });
    } catch (e) { p.errors[platform] = e.message; log({ action: "failed", platform, post: p.id, client: p.client, error: e.message }); }
  }
  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;
      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 };
}