[object Object]

← back to Marketing Command Center

engine: core module (store/config/skeleton) + channels.postLive + leak-deny lib

0938d51debc27c48c2fbb82d1de93bdbbfab55ff · 2026-07-21 14:36:07 -0700 · Steve

Files touched

Diff

commit 0938d51debc27c48c2fbb82d1de93bdbbfab55ff
Author: Steve <steve@designerwallcoverings.com>
Date:   Tue Jul 21 14:36:07 2026 -0700

    engine: core module (store/config/skeleton) + channels.postLive + leak-deny lib
---
 data/engine-queue.json    |   1 +
 lib/leak-deny.js          |  28 ++++++++
 modules/channels/index.js |  18 +++++
 modules/engine/config.js  |  78 +++++++++++++++++++++
 modules/engine/index.js   | 172 ++++++++++++++++++++++++++++++++++++++++++++++
 modules/engine/store.js   |  82 ++++++++++++++++++++++
 modules/registry.js       |   1 +
 7 files changed, 380 insertions(+)

diff --git a/data/engine-queue.json b/data/engine-queue.json
new file mode 100644
index 0000000..0637a08
--- /dev/null
+++ b/data/engine-queue.json
@@ -0,0 +1 @@
+[]
\ No newline at end of file
diff --git a/lib/leak-deny.js b/lib/leak-deny.js
new file mode 100644
index 0000000..10a9169
--- /dev/null
+++ b/lib/leak-deny.js
@@ -0,0 +1,28 @@
+// Private-label LEAK guard — shared across the marketing stack so no upstream/true
+// vendor name ever surfaces customer-facing (captions, hashtags, product titles).
+// These are private-label lines whose real supplier must NEVER appear publicly.
+// Mirrors dw-marketing-reels/scripts/build-reel.mjs (LEAK_DENY + clean()) and the
+// dw-leak-scanner denylist. Copy of the reel's regex + clean() approach.
+//
+// Usage:
+//   const { LEAK_DENY, clean, isLeaky } = require('../lib/leak-deny');
+//   if (isLeaky(caption)) …           // does ANY denylisted name appear?
+//   const safeCaption = clean(caption); // strip any denylisted word out
+
+// Denylist regex — the upstream vendor names that must stay private-label.
+const LEAK_DENY = /greenland|wallquest|chesapeake|nextwall|seabrook|brewster|command\s*54|momentum|versa|desima|carlsten|nicolette\s*mayer|york\s*wall/i;
+
+// isLeaky(s) — true if the string contains any denylisted upstream name.
+function isLeaky(s) {
+  return !!s && LEAK_DENY.test(String(s));
+}
+
+// clean(s) — final safety net: drop any whitespace-delimited word that matches
+// the denylist, so a single leaked token can't ride along in an otherwise-safe
+// string. Returns the scrubbed string (empty string for null/undefined input).
+function clean(s) {
+  if (s == null) return '';
+  return String(s).split(/\s+/).filter(w => !LEAK_DENY.test(w)).join(' ');
+}
+
+module.exports = { LEAK_DENY, clean, isLeaky };
diff --git a/modules/channels/index.js b/modules/channels/index.js
index 8ef352d..d3f4eea 100644
--- a/modules/channels/index.js
+++ b/modules/channels/index.js
@@ -851,3 +851,21 @@ module.exports.redirectUriFor = redirectUri;
 // (a dead Meta token flips connected:false instead of showing green on env-presence).
 module.exports.metaTokenHealth = metaTokenHealth;
 module.exports.augmentMetaHealth = augmentMetaHealth;
+
+// Programmatic live publish for the engine scheduler. Approval in the Engine
+// panel IS the confirm; this enforces the same rails as /publish.
+module.exports.postLive = async function postLive(channel, content, opts = {}) {
+  if (!ADAPTERS[channel]) return { ok: false, error: 'unknown channel' };
+  if (!content.mediaUrl && !content.videoUrl)
+    return { ok: false, error: 'media required — DW never posts text-only' };
+  const st = platformStatus();
+  if (!st[channel] || !st[channel].connected) return { ok: false, error: 'not connected' };
+  const r = await ADAPTERS[channel](content, opts).catch(e => [{ ok: false, error: e.message }]);
+  const ok = Array.isArray(r) && r.length > 0 && r.every(x => x.ok);
+  const outbox = readOutbox();
+  const entry = { id: 'eng-' + channel + '-' + Date.now().toString(36), channel,
+    caption: content.caption, mediaUrl: content.mediaUrl, status: ok ? 'posted' : 'failed',
+    detail: r, via: 'engine', at: new Date().toISOString() };
+  outbox.push(entry); writeOutbox(outbox);
+  return { ok, detail: r, outboxId: entry.id };
+};
diff --git a/modules/engine/config.js b/modules/engine/config.js
new file mode 100644
index 0000000..474e411
--- /dev/null
+++ b/modules/engine/config.js
@@ -0,0 +1,78 @@
+// Engine config — defaults + persisted overrides for the daily-suggestion engine.
+// getConfig() returns DEFAULTS merged with data/engine-config.json; setConfig()
+// shallow-merges a patch and persists it atomically (tmp → rename).
+const fs = require('fs');
+const path = require('path');
+
+const CONFIG_FILE = path.join(__dirname, '..', '..', 'data', 'engine-config.json');
+
+// Per-channel default posting slots (local HH:MM). Approve defaults an item to
+// today at its suggestedSlot (drawn from these).
+const DEFAULTS = {
+  slots: {
+    instagram: ['11:00', '17:00'],
+    facebook: ['09:00', '15:00'],
+    bluesky: ['10:00'],
+    youtube: ['12:00'],
+    tiktok: ['18:00'],
+    threads: ['12:30'],
+  },
+  productsPerDay: 3,   // how many products the generator picks per day
+  lookbackDays: 14,    // how far back to pull candidate products
+  noRepeatDays: 14,    // don't re-feature the same product within this window
+};
+
+// Per-channel caption character ceilings (used by later agents to trim/validate).
+const CHANNEL_LIMITS = {
+  bluesky: 280,
+  threads: 480,
+  instagram: 2200,
+  facebook: 2200,
+  tiktok: 150,
+  youtube: 95,
+};
+
+// Channels held behind platform App Review — the generator marks these items
+// status:'gated' with the reason below until /arm flips them live.
+const GATED_CHANNELS = {
+  tiktok: 'App Review pending — posts private-only (SELF_ONLY)',
+  threads: 'App Review pending — live publish requires Meta approval',
+};
+
+function readOverrides() {
+  try { return JSON.parse(fs.readFileSync(CONFIG_FILE, 'utf8')); }
+  catch { return {}; }
+}
+
+function writeOverrides(obj) {
+  try { fs.mkdirSync(path.dirname(CONFIG_FILE), { recursive: true }); } catch {}
+  const tmp = CONFIG_FILE + '.tmp';
+  fs.writeFileSync(tmp, JSON.stringify(obj, null, 2));
+  fs.renameSync(tmp, CONFIG_FILE);
+}
+
+// getConfig() — DEFAULTS with persisted overrides merged over (slots merged one
+// level deep so a partial slots override doesn't wipe unspecified channels).
+function getConfig() {
+  const o = readOverrides();
+  return {
+    ...DEFAULTS,
+    ...o,
+    slots: { ...DEFAULTS.slots, ...(o.slots || {}) },
+  };
+}
+
+// setConfig(patch) — shallow-merge patch into the stored overrides, persist,
+// and return the effective merged config.
+function setConfig(patch = {}) {
+  const o = readOverrides();
+  const next = {
+    ...o,
+    ...patch,
+    slots: { ...(o.slots || {}), ...(patch.slots || {}) },
+  };
+  writeOverrides(next);
+  return getConfig();
+}
+
+module.exports = { DEFAULTS, CHANNEL_LIMITS, GATED_CHANNELS, getConfig, setConfig };
diff --git a/modules/engine/index.js b/modules/engine/index.js
new file mode 100644
index 0000000..5a0dc88
--- /dev/null
+++ b/modules/engine/index.js
@@ -0,0 +1,172 @@
+// Engine — daily AI-suggested social posts with an approve → schedule → post flow.
+// This module owns the SUGGESTION QUEUE (data/engine-queue.json via store.js). Each
+// day the generator (agent 2 — ./generate) proposes a few product posts per channel;
+// Steve approves/skips/edits them here; the scheduler (agent 3 — ./scheduler) fires
+// approved items at their scheduled time through channels.postLive (the same rails
+// as the /publish route). This foundation module is safe with ./generate and
+// ./scheduler ABSENT — both are lazy-required in try/catch so a missing file never
+// crashes the mount.
+const store = require('./store');
+const config = require('./config');
+const { clean } = require('../../lib/leak-deny');
+
+// ── local-date helper ────────────────────────────────────────────────────────
+// The queue keys suggestions on the local calendar day. Format YYYY-MM-DD in the
+// server's local timezone (NOT UTC — toISOString would drift across midnight).
+function today() {
+  const d = new Date();
+  const p = n => String(n).padStart(2, '0');
+  return `${d.getFullYear()}-${p(d.getMonth() + 1)}-${p(d.getDate())}`;
+}
+
+// Lazily load the generator (agent 2 builds ./generate). Absence is fine.
+function loadGenerator() {
+  let gen;
+  try { gen = require('./generate'); } catch { gen = null; }
+  return gen;
+}
+
+// Build a full ISO string (with local offset) for `date` at `HH:MM` local time.
+// If that time has already passed today, roll to the same slot tomorrow.
+function slotToISO(date, hhmm, { rollIfPast = false } = {}) {
+  const [h, m] = String(hhmm || '00:00').split(':').map(Number);
+  const d = new Date(`${date}T00:00:00`);
+  d.setHours(h || 0, m || 0, 0, 0);
+  if (rollIfPast && d.getTime() <= Date.now()) d.setDate(d.getDate() + 1);
+  return d.toISOString();
+}
+
+module.exports = {
+  id: 'engine',
+  title: 'Engine',
+  icon: '⚙️',
+
+  mount(router) {
+    // Kick the scheduler if agent 3 has built it. Absence must NOT crash mount.
+    try {
+      const scheduler = require('./scheduler');
+      if (scheduler && typeof scheduler.start === 'function') scheduler.start();
+    } catch { /* scheduler not built yet — fine */ }
+
+    // GET /status — engine + queue health for today.
+    router.get('/status', (_req, res) => {
+      const items = store.load();
+      const d = today();
+      const todays = items.filter(it => it.date === d);
+      const counts = { suggested: 0, approved: 0, posting: 0, posted: 0, failed: 0, skipped: 0, gated: 0 };
+      for (const it of todays) if (counts[it.status] != null) counts[it.status]++;
+      res.json({
+        scheduler: process.env.ENGINE_SCHEDULER !== 'off',
+        lastTick: null, // scheduler agent will wire the real last-tick timestamp
+        counts,
+        total: items.length,
+      });
+    });
+
+    // GET /today — today's suggestions. If none exist and generate isn't disabled
+    // (?generate=0), ask the generator to build them. Missing generator → empty.
+    router.get('/today', async (req, res) => {
+      const d = today();
+      let items = store.load().filter(it => it.date === d);
+      if (!items.length && req.query.generate !== '0') {
+        const gen = loadGenerator();
+        if (!gen) return res.json({ items: [], note: 'generator not built yet' });
+        try {
+          await gen.generateToday();
+          items = store.load().filter(it => it.date === d);
+        } catch (e) {
+          return res.status(500).json({ items: [], error: e.message });
+        }
+      }
+      res.json({ items });
+    });
+
+    // POST /generate — force a generation run now (same lazy require).
+    router.post('/generate', async (_req, res) => {
+      const gen = loadGenerator();
+      if (!gen) return res.json({ items: [], note: 'generator not built yet' });
+      try {
+        await gen.generateToday();
+        const items = store.load().filter(it => it.date === today());
+        res.json({ items });
+      } catch (e) {
+        res.status(500).json({ error: e.message });
+      }
+    });
+
+    // POST /approve { id, scheduledAt?, privacy? } — suggested → approved.
+    // Defaults scheduledAt to today at the item's suggestedSlot (rolling to
+    // tomorrow's same slot if that time already passed). YouTube defaults to
+    // privacy 'public' unless the caller specifies one.
+    router.post('/approve', (req, res) => {
+      const { id, scheduledAt, privacy } = req.body || {};
+      if (!id) return res.status(400).json({ ok: false, error: 'id required' });
+      const it = store.get(id);
+      if (!it) return res.status(404).json({ ok: false, error: 'item not found' });
+
+      const patch = {};
+      patch.scheduledAt = scheduledAt || slotToISO(today(), it.suggestedSlot, { rollIfPast: true });
+      if (it.channel === 'youtube') patch.privacy = privacy || 'public';
+      else if (privacy != null) patch.privacy = privacy;
+
+      const updated = store.transition(id, 'suggested', 'approved', patch);
+      if (!updated) return res.status(409).json({ ok: false, error: `cannot approve — item is not 'suggested'` });
+      res.json({ ok: true, item: updated });
+    });
+
+    // POST /skip { id } — suggested → skipped.
+    router.post('/skip', (req, res) => {
+      const { id } = req.body || {};
+      if (!id) return res.status(400).json({ ok: false, error: 'id required' });
+      const updated = store.transition(id, 'suggested', 'skipped', {});
+      if (!updated) return res.status(409).json({ ok: false, error: `cannot skip — item is not 'suggested'` });
+      res.json({ ok: true, item: updated });
+    });
+
+    // POST /arm { id } — gated → suggested. For when TikTok/Threads App Review lands
+    // and a held item can re-enter the normal approve flow.
+    router.post('/arm', (req, res) => {
+      const { id } = req.body || {};
+      if (!id) return res.status(400).json({ ok: false, error: 'id required' });
+      const updated = store.transition(id, 'gated', 'suggested', { gateReason: null });
+      if (!updated) return res.status(409).json({ ok: false, error: `cannot arm — item is not 'gated'` });
+      res.json({ ok: true, item: updated });
+    });
+
+    // POST /edit { id, caption?, hashtags?, mediaUrl?, scheduledAt? } — only while
+    // suggested|approved. Re-runs the leak clean() on any edited caption.
+    router.post('/edit', (req, res) => {
+      const { id, caption, hashtags, mediaUrl, scheduledAt } = req.body || {};
+      if (!id) return res.status(400).json({ ok: false, error: 'id required' });
+      const it = store.get(id);
+      if (!it) return res.status(404).json({ ok: false, error: 'item not found' });
+      if (it.status !== 'suggested' && it.status !== 'approved') {
+        return res.status(409).json({ ok: false, error: `cannot edit — status is '${it.status}'` });
+      }
+      const patch = {};
+      if (caption != null) patch.caption = clean(caption);
+      if (hashtags != null) patch.hashtags = Array.isArray(hashtags) ? hashtags : it.hashtags;
+      if (mediaUrl != null) patch.mediaUrl = mediaUrl;
+      if (scheduledAt != null) patch.scheduledAt = scheduledAt;
+      const updated = store.upsert({ ...it, ...patch });
+      res.json({ ok: true, item: updated });
+    });
+
+    // GET /queue?status=&days=7 — filtered history, newest first.
+    router.get('/queue', (req, res) => {
+      const { status } = req.query;
+      const days = Math.max(1, parseInt(req.query.days, 10) || 7);
+      const cutoff = new Date();
+      cutoff.setDate(cutoff.getDate() - days);
+      const cutoffDay = cutoff.toISOString().slice(0, 10);
+      let items = store.load().filter(it => (it.date || '') >= cutoffDay);
+      if (status) items = items.filter(it => it.status === status);
+      items.sort((a, b) => String(b.createdAt || '').localeCompare(String(a.createdAt || '')));
+      res.json({ items });
+    });
+
+    // GET /config — effective config. PUT /config — merge + persist a patch.
+    router.get('/config', (_req, res) => res.json(config.getConfig()));
+    router.put('/config', (req, res) => res.json(config.setConfig(req.body || {})));
+  },
+};
diff --git a/modules/engine/store.js b/modules/engine/store.js
new file mode 100644
index 0000000..f328704
--- /dev/null
+++ b/modules/engine/store.js
@@ -0,0 +1,82 @@
+// Engine queue store — persistence for data/engine-queue.json (an array of items).
+// Atomic writes (write tmp → fs.renameSync) so a crash mid-write never leaves a
+// half-written / corrupt queue (mirrors server.js writeNotes()). load() returns []
+// on missing/corrupt so a bad file never crashes the module.
+//
+// QUEUE ITEM SHAPE (later agents — candidates.js / generate.js / scheduler.js —
+// MUST conform to this exactly):
+// {
+//   id: 'eng-<yyyymmdd>-<dwSkuOrShopifyId>-<channel>', // stable, dedupe key
+//   date: 'YYYY-MM-DD',           // the suggestion's local calendar day
+//   createdAt: ISO,
+//   product: {
+//     shopifyId, dwSku, handle, title, vendor, productType, imageUrl, productUrl
+//   },
+//   channel,                      // instagram|facebook|bluesky|youtube|tiktok|threads
+//   caption,
+//   hashtags: [],
+//   mediaUrl,
+//   videoUrl: null,
+//   suggestedSlot: 'HH:MM',       // from config.slots[channel]
+//   scheduledAt: null|ISO,        // set on approve (full ISO w/ offset)
+//   privacy: null,                // youtube: 'public'|'private'|'unlisted'
+//   status: 'suggested'|'approved'|'posting'|'posted'|'failed'|'skipped'|'gated',
+//   gateReason: null,             // why a 'gated' item is held (TikTok/Threads review)
+//   source: { llm, cost: 0 },     // provenance from the LLM that wrote the caption
+//   attempts: 0,
+//   lastError: null,
+//   outboxId: null,               // channels-outbox id once posted
+//   postedAt: null,
+//   detail: null                  // adapter result on post
+// }
+const fs = require('fs');
+const path = require('path');
+
+const QUEUE_FILE = path.join(__dirname, '..', '..', 'data', 'engine-queue.json');
+
+// load() — return the queue array, or [] on missing/corrupt (never throws).
+function load() {
+  try {
+    const v = JSON.parse(fs.readFileSync(QUEUE_FILE, 'utf8'));
+    return Array.isArray(v) ? v : [];
+  } catch { return []; }
+}
+
+// save(items) — atomic write (tmp → rename) of the whole queue array.
+function save(items) {
+  try { fs.mkdirSync(path.dirname(QUEUE_FILE), { recursive: true }); } catch {}
+  const tmp = QUEUE_FILE + '.tmp';
+  fs.writeFileSync(tmp, JSON.stringify(Array.isArray(items) ? items : [], null, 2));
+  fs.renameSync(tmp, QUEUE_FILE);
+}
+
+// get(id) — one item by id, or null.
+function get(id) {
+  return load().find(it => it.id === String(id)) || null;
+}
+
+// upsert(item) — add or replace the item with the same id, persist, return it.
+function upsert(item) {
+  const items = load();
+  const i = items.findIndex(x => x.id === item.id);
+  if (i >= 0) items[i] = { ...items[i], ...item };
+  else items.push(item);
+  save(items);
+  return item;
+}
+
+// transition(id, fromStatus, toStatus, patch={}) — the compare-and-set that
+// guards against double-posting. Re-reads the file FRESH, finds the item, asserts
+// its current status === fromStatus (returns null if not — the guard), applies
+// { status: toStatus, ...patch }, persists, and returns the updated item.
+function transition(id, fromStatus, toStatus, patch = {}) {
+  const items = load();
+  const i = items.findIndex(x => x.id === String(id));
+  if (i < 0) return null;
+  if (items[i].status !== fromStatus) return null; // compare-and-set failed
+  items[i] = { ...items[i], ...patch, status: toStatus };
+  save(items);
+  return items[i];
+}
+
+module.exports = { load, save, get, upsert, transition };
diff --git a/modules/registry.js b/modules/registry.js
index c95655d..fbbd272 100644
--- a/modules/registry.js
+++ b/modules/registry.js
@@ -34,4 +34,5 @@ module.exports = [
   'journeys',           // cross-channel journeys: email → wait → SMS/social → retarget per segment, persisted, with recommendations + dry-run (STAGED)
   'browse-abandon',     // browse-abandonment drip: Day1 storytelling → Day14 room-setting → Day30 suppress, mock events + dry-run preview (STAGED)
   'playbook',           // monthly playbook: 4-6 suggested campaigns/actions for next 4 weeks from calendar + segment health + perf gaps (Gemini or template)
+  'engine',             // Engine: daily AI-suggested social posts with an approve → schedule → post flow (queue via store.js, live via channels.postLive)
 ];

← e8501e8 assets: local qwen2.5vl vision enrichment (aiTags) + search  ·  back to Marketing Command Center  ·  engine: brain (candidates+llm+generate), scheduler (CAS tick 0b73831 →