// 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://instantadpay.com'; // 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) return { ok: true, already: true }; await impl().update(e, { stopped: 2 }); return { ok: true }; } // ---- 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 InstantAdPay account. Stop them here: ' + unsubUrl(email) + '\n\nInstantAdPay ยท 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, DEFAULT_SEQUENCE };