linkspin/drip.js

// Follow-up email sequence for new free members ("the lead came in the door
// with their email"). A row is queued when an account is created; a ticker
// sends each step when it comes due; a signed unsubscribe link stops it.
// Sequence copy lives in DATA_DIR/drip.json (admin-editable) with the
// defaults below. Dual-mode storage like the rest of the site (MySQL / JSON).
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const db = require('./db');

let DATA_DIR = null, mailer = null, accounts = null, chain = null, SITE = 'https://linkspin-test.saasy.top';

// hours = time after the account was created. Copy drafted through Marty's
// Branded Voice engine (one framework per email: gain, logic, PAS, logic,
// honest fear of loss, AIDA, gain + hand-off to the newsletter), then checked
// line by line against the contract facts. Admin overrides live in drip.json.
const DEFAULT_SEQUENCE = require('./drip-defaults.json');

// ---- storage ----
const J = {
  db: null,
  FILE: () => path.join(DATA_DIR, 'drips.json'),
  load() { try { this.db = JSON.parse(fs.readFileSync(this.FILE(), 'utf8')); } catch (e) { this.db = {}; } },
  save() { try { fs.writeFileSync(this.FILE(), JSON.stringify(this.db)); } catch (e) {} },
  async enqueue(rec) { if (!this.db) this.load(); if (this.db[rec.email]) return false; this.db[rec.email] = rec; this.save(); return true; },
  async due(now, limit) { if (!this.db) this.load(); return Object.values(this.db).filter(r => !r.stopped && r.nextAt <= now).sort((a, b) => a.nextAt - b.nextAt).slice(0, limit); },
  async update(email, fields) { if (!this.db) this.load(); if (this.db[email]) { Object.assign(this.db[email], fields); this.save(); } },
  async stats() { if (!this.db) this.load(); const v = Object.values(this.db); return { active: v.filter(r => !r.stopped).length, unsubscribed: v.filter(r => r.stopped === 2).length, done: v.filter(r => r.stopped === 1).length, total: v.length }; },
  async get(email) { if (!this.db) this.load(); return this.db[email] || null; }
};
const rowR = r => ({ email: r.email, step: r.step, nextAt: Number(r.next_at), started: Number(r.started), stopped: r.stopped, ref: r.ref, angle: r.angle });
const D = {
  async enqueue(rec) {
    try {
      await db.q('INSERT INTO drips (email,step,next_at,started,stopped,ref,angle) VALUES (?,?,?,?,0,?,?)',
        [rec.email, rec.step, rec.nextAt, rec.started, rec.ref || null, rec.angle || null]);
      return true;
    } catch (e) { if (e.code === 'ER_DUP_ENTRY') return false; throw e; }
  },
  async due(now, limit) { return (await db.q('SELECT * FROM drips WHERE stopped=0 AND next_at<=? ORDER BY next_at LIMIT ?', [now, limit])).map(rowR); },
  async update(email, fields) {
    const sets = [], vals = [];
    if ('step' in fields) { sets.push('step=?'); vals.push(fields.step); }
    if ('nextAt' in fields) { sets.push('next_at=?'); vals.push(fields.nextAt); }
    if ('stopped' in fields) { sets.push('stopped=?'); vals.push(fields.stopped); }
    if (!sets.length) return;
    vals.push(email);
    await db.q('UPDATE drips SET ' + sets.join(',') + ' WHERE email=?', vals);
  },
  async stats() {
    const r = await db.q('SELECT SUM(stopped=0) active, SUM(stopped=2) unsubscribed, SUM(stopped=1) done, COUNT(*) total FROM drips');
    return { active: Number(r[0].active || 0), unsubscribed: Number(r[0].unsubscribed || 0), done: Number(r[0].done || 0), total: Number(r[0].total || 0) };
  },
  async get(email) { const r = await db.q('SELECT * FROM drips WHERE email=?', [email]); return r.length ? rowR(r[0]) : null; }
};
const impl = () => db.enabled() ? D : J;

// ---- sequence config ----
function seqFile() { return path.join(DATA_DIR, 'drip.json'); }
function sequence() {
  try {
    const saved = JSON.parse(fs.readFileSync(seqFile(), 'utf8'));
    if (Array.isArray(saved) && saved.length) return saved.map(normStep).filter(Boolean);
  } catch (e) {}
  return DEFAULT_SEQUENCE;
}
function normStep(s) {
  if (!s || typeof s !== 'object') return null;
  const hours = Number(s.hours);
  const subject = String(s.subject || '').trim().slice(0, 150);
  const body = String(s.body || '').trim().slice(0, 8000);
  if (!(hours >= 1) || !subject || !body) return null;
  return { hours, subject, body };
}
function setSequence(arr) {
  if (!Array.isArray(arr)) return { error: 'Send a list of steps.' };
  const steps = arr.map(normStep);
  if (steps.some(s => !s)) return { error: 'Every step needs hours (>= 1), a subject and a body.' };
  if (steps.length > 20) return { error: 'Keep it to 20 steps or fewer.' };
  for (let i = 1; i < steps.length; i++) if (steps[i].hours <= steps[i - 1].hours) return { error: 'Steps must be in increasing hours.' };
  fs.writeFileSync(seqFile(), JSON.stringify(steps, null, 2));
  return { ok: true, sequence: steps };
}
function resetSequence() { try { fs.unlinkSync(seqFile()); } catch (e) {} return { ok: true, sequence: DEFAULT_SEQUENCE }; }

// ---- unsubscribe signing ----
function secret() {
  const f = path.join(DATA_DIR, 'drip.secret');
  try { return fs.readFileSync(f, 'utf8').trim(); } catch (e) {}
  const s = crypto.randomBytes(24).toString('hex');
  try { fs.writeFileSync(f, s, { mode: 0o600 }); } catch (e) {}
  return s;
}
function token(email) { return crypto.createHmac('sha256', secret()).update(String(email).toLowerCase()).digest('hex').slice(0, 32); }
function unsubUrl(email) { return SITE + '/unsubscribe?e=' + encodeURIComponent(email) + '&t=' + token(email); }
async function unsubscribe(email, t) {
  const e = String(email || '').trim().toLowerCase();
  if (!e || !t || t !== token(e)) return { error: 'That link is not valid.' };
  const cur = await impl().get(e);
  if (!cur) { try { await enqueue(e, '', ''); await impl().update(e, { stopped: 2 }); } catch (err) {} return { ok: true }; } // member without a drip row: record the opt-out anyway (update emails honour it)
  await impl().update(e, { stopped: 2 });
  return { ok: true };
}
async function isUnsubscribed(email) { const cur = await impl().get(String(email || '').toLowerCase()); return !!(cur && cur.stopped === 2); }

// ---- rendering ----
async function vars(email) {
  const a = accounts ? await accounts.byEmail(email) : null;
  const tok = a ? (a.username || a.code || (a.memberId ? String(a.memberId) : '')) : '';
  let sponsor = 'your sponsor';
  try {
    const sp = accounts && await accounts.sponsorOf(email);
    if (sp) sponsor = sp.username ? '@' + sp.username : (sp.memberId ? 'member #' + sp.memberId : 'your sponsor');
  } catch (e) {}
  // paid = has bought at least one package (on-chain credits minted); used by {{paid:a|b}}
  let paid = false;
  try { if (a && a.memberId && chain) paid = (await chain.creditBalance(a.memberId, 0)) > 0; } catch (e) {}
  return {
    link: tok ? SITE + '/join/' + tok : SITE + '/my',
    sponsor, site: SITE, email, paid,
    footer: 'You are getting these follow-ups because you created a free LinkSpin account. Stop them here: ' + unsubUrl(email)
      + '\n\nLinkSpin · Advertising, not investing. No income is guaranteed; results depend on your effort. Crypto carries risk of loss.'
  };
}
// simple {{key}} placeholders resolve first (they may sit inside a branch),
// then {{paid:words if they bought|words if not}} picks the branch
function render(text, v) {
  return String(text)
    .replace(/\{\{(\w+)\}\}/g, (m, k) => (k in v && typeof v[k] !== 'boolean' ? v[k] : m))
    .replace(/\{\{paid:([\s\S]*?)\|([\s\S]*?)\}\}/g, (m, yes, no) => (v.paid ? yes : no));
}

// ---- lifecycle ----
function init(opts) {
  DATA_DIR = opts.dataDir; mailer = opts.mailer; accounts = opts.accounts; chain = opts.chain || null;
  if (opts.site) SITE = opts.site;
}
// queue a fresh account; step 0 of the sequence is due `hours` after creation
async function enqueue(email, ref, angle) {
  const e = String(email || '').trim().toLowerCase();
  if (!e) return false;
  const seq = sequence();
  const now = Date.now();
  return impl().enqueue({ email: e, step: 0, nextAt: now + seq[0].hours * 3600000, started: now, stopped: 0,
    ref: String(ref || '').slice(0, 40) || null, angle: String(angle || '').slice(0, 20) || null });
}
async function sendStep(email, stepIdx, to) {
  const seq = sequence();
  const s = seq[stepIdx];
  if (!s) return { error: 'No such step.' };
  const v = await vars(email);
  await mailer.send(to || email, render(s.subject, v), render(s.body, v));
  return { ok: true };
}
let ticking = false;
async function tick() {
  if (ticking || !mailer || !mailer.hasKey()) return 0;
  ticking = true;
  let sent = 0;
  try {
    const seq = sequence();
    const now = Date.now();
    const rows = await impl().due(now, 50);
    for (const r of rows) {
      try {
        if (r.step >= seq.length) { await impl().update(r.email, { stopped: 1 }); continue; }
        await sendStep(r.email, r.step);
        sent += 1;
        const next = r.step + 1;
        if (next >= seq.length) await impl().update(r.email, { step: next, stopped: 1 });
        else await impl().update(r.email, { step: next, nextAt: Math.max(now + 60000, r.started + seq[next].hours * 3600000) });
      } catch (e) {
        console.error('drip send', r.email, e.message);
        await impl().update(r.email, { nextAt: now + 6 * 3600000 }); // retry later, do not spin
      }
    }
  } finally { ticking = false; }
  return sent;
}
async function stats() { return impl().stats(); }

module.exports = { init, enqueue, tick, stats, sequence, setSequence, resetSequence, sendStep, unsubscribe, unsubUrl, isUnsubscribed, DEFAULT_SEQUENCE };