instantadpay/pipeline.js

// Pipeline (Marty, 2026-09-15): a follow-up board for sponsors. One screen where every prospect
// and every direct sits in a column by what they have actually done. Nobody drags a card: the
// site moves it when the event fires (wallet linked, payouts on, first buy, buyers counted).
// The sponsor adds what the site cannot know: a note, a follow-up date, an outcome tag.
// Built before launch, shown as "coming soon" until site setting pipelineMode = on
// (preview = only the admin account sees the live board, for testing and the training video).
// Dual-mode store like the other modules (MySQL when DATABASE_URL is set, JSON on the volume).
const fs = require('fs');
const path = require('path');
const db = require('./db');

let DATA_DIR = null, accounts = null, coach = null, chain = null;
// linked positions (Qualified Start) belong to the member and their buyers count toward badges, but the contract
// pays level 2 and 3 to the MAIN wallet only when that wallet's own buyerCount reaches 2 and 5. The card shows both
// numbers; the stage and the advice follow the main wallet, because that is what gets paid (Marty's audit, 2026-09-15)
const posCache = new Map();
async function positionBuyers(email) {
  let ids = []; try { ids = (await accounts.positions(email)).map(p => p.memberId).filter(Boolean); } catch (e) {}
  let n = 0;
  for (const id of ids) {
    const c = posCache.get(id); let v = c && Date.now() - c.t < 60000 ? c.v : null;
    if (v == null) { try { v = (await chain.member(id)).buyerCount || 0; } catch (e) { v = 0; } posCache.set(id, { t: Date.now(), v }); }
    n += v;
  }
  return { count: n, positions: ids.length };
}
const DAY = 86400000;

// columns, in order. Prospects live in the first and last; directs land by coaching rung.
const STAGES = [
  { key: 'talking', label: 'Talking to', hint: 'People you have reached out to who have not joined yet.' },
  { key: 'joined', label: 'Joined', hint: 'Joined free. Next: link a wallet.' },
  { key: 'wallet', label: 'Wallet linked', hint: 'Next: switch on payouts, one free transaction.' },
  { key: 'payouts', label: 'Payouts on', hint: 'Next: a first package. $20 or more counts for you.' },
  { key: 'bought', label: 'Bought', hint: 'Your qualifying buyer. Next: their first person.' },
  { key: 'building', label: 'Building', hint: 'They have buyers of their own. Coach them to their two.' },
  { key: 'later', label: 'Later', hint: 'Parked: not now, no response, not interested.' }
];
const TAGS = ['', 'hot', 'later', 'no response', 'not interested'];
const RUNG_STAGE = ['joined', 'wallet', 'payouts', 'bought', 'building', 'building', 'building'];
const PARKED = new Set(['later', 'no response', 'not interested']);

// ---- stores: one row per (owner, person) ----
const J = {
  db: null,
  FILE: () => path.join(DATA_DIR, 'pipeline.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 notes(owner) { if (!this.db) this.load(); return Object.values(this.db).filter(r => r.owner === owner); },
  async put(owner, person, f) { if (!this.db) this.load(); const k = owner + '|' + person; this.db[k] = Object.assign(this.db[k] || { owner, person }, f, { updated: Date.now() }); this.save(); return this.db[k]; }
};
const rowR = r => ({ owner: r.owner_email, person: r.person, note: r.note || '', followUp: r.follow_up ? Number(r.follow_up) : null, tag: r.tag || '', updated: Number(r.updated) });
const D = {
  async notes(owner) { return (await db.q('SELECT * FROM pipeline_notes WHERE owner_email=?', [owner])).map(rowR); },
  async put(owner, person, f) {
    await db.q('INSERT INTO pipeline_notes (owner_email,person,note,follow_up,tag,updated) VALUES (?,?,?,?,?,?) ON DUPLICATE KEY UPDATE note=VALUES(note), follow_up=VALUES(follow_up), tag=VALUES(tag), updated=VALUES(updated)',
      [owner, person, f.note, f.followUp || null, f.tag || '', Date.now()]);
    return rowR((await db.q('SELECT * FROM pipeline_notes WHERE owner_email=? AND person=?', [owner, person]))[0]);
  }
};
const impl = () => db.enabled() ? D : J;

function init(opts) { DATA_DIR = opts.dataDir; accounts = opts.accounts; coach = opts.coach; chain = opts.chain; }

// who may see the live board: mode on = everyone; preview = the admin account only
function visible(mode, email, adminEmail) {
  if (mode === 'on') return true;
  if (mode === 'preview') return !!(email && adminEmail && String(email).toLowerCase() === String(adminEmail).toLowerCase());
  return false;
}

async function board(email) {
  const owner = String(email || '').toLowerCase();
  const now = Date.now(); const eod = new Date(); eod.setHours(23, 59, 59, 999);
  const notes = {}; for (const n of await impl().notes(owner)) notes[n.person] = n;
  const cards = [];
  // prospects: the sponsor's own list of people they have talked to
  for (const p of await coach.prospects(owner)) {
    const n = notes['p:' + p.id] || {};
    let stage = 'talking';
    if (p.status === 'not now') stage = 'later';
    if (p.status === 'joined') stage = 'joined';
    if (p.status === 'bought') stage = 'bought';
    if (PARKED.has(n.tag)) stage = 'later';
    cards.push({ key: 'p:' + p.id, kind: 'prospect', name: p.name, contact: p.contact || '', status: p.status, stage,
      since: p.updated || p.created || now, lastSeen: 0, quietDays: Math.floor((now - (p.updated || p.created || now)) / DAY),
      stalled: false, next: p.status === 'contacted' || p.status === 'interested' ? 'Send your link and ask for a yes' : 'Reach out once',
      say: 'Hey ' + p.name + ', here is the link I mentioned. Free to join with an email, and I will walk you through the first three steps: {{link}}',
      note: n.note || p.note || '', followUp: n.followUp || p.nextTs || null, tag: n.tag || '' });
  }
  // directs: stage from what they have actually done (the coaching rung)
  const levels = await accounts.downline(owner, 1);
  for (const m of (levels[0] ? levels[0].members : [])) {
    let c = null; try { c = await coach.describe(m, now); } catch (e) {}
    const n = notes[m.email] || {};
    const pb = c && c.rung >= 3 ? await positionBuyers(m.email) : { count: 0, positions: 0 };
    const buyersMain = c ? c.buyerCount || 0 : 0, buyersAll = buyersMain + pb.count;
    const rung = c ? c.rung : 0;
    const RR = coach.RUNGS[rung];
    // when positions carry buyers the main wallet does not, say exactly what the contract needs
    const gap = pb.count && rung >= 3 && rung < 6 ? ' The contract pays level ' + (buyersMain < 2 ? '2' : '3') + ' to their main wallet only when that wallet has ' + (buyersMain < 2 ? 2 : 5) + ' qualifying buyers of its own: ' + buyersMain + ' now, plus ' + pb.count + ' on linked positions that count for badges only.' : '';
    const stage = PARKED.has(n.tag) && rung < 3 ? 'later' : RUNG_STAGE[rung] || 'joined';
    cards.push({ key: m.email, kind: 'member', email: m.email, name: m.username ? '@' + m.username : (m.memberId ? 'member #' + m.memberId : 'member'),
      memberId: m.memberId || 0, stage, rung, rungLabel: RR.label, since: m.created || now,
      lastSeen: m.lastSeen || 0, quietDays: c ? c.quietDays : 0, stalled: !!(c && c.stalled && rung < 6),
      buyers: buyersAll, buyersMain, buyersPositions: pb.count, positions: pb.positions, bought: !!(c && c.counted),
      next: RR.next + gap, say: RR.say.replace('{{name}}', m.username || 'there'),
      note: n.note || '', followUp: n.followUp || null, tag: n.tag || '' });
  }
  const due = cards.filter(c => c.followUp && c.followUp <= eod.getTime()).sort((a, b) => a.followUp - b.followUp);
  const columns = STAGES.map(s => ({ key: s.key, label: s.label, hint: s.hint,
    cards: cards.filter(c => c.stage === s.key).sort((a, b) => (b.stalled - a.stalled) || (a.followUp || 9e15) - (b.followUp || 9e15) || b.since - a.since) }));
  return { stages: STAGES, tags: TAGS, columns, due, counts: { total: cards.length, stalled: cards.filter(c => c.stalled).length, due: due.length } };
}

async function save(email, body) {
  const owner = String(email || '').toLowerCase();
  const person = String(body.key || '').trim().toLowerCase().slice(0, 190);
  if (!person) return { error: 'Pick who to update.' };
  // only people on the sponsor's own board
  const ok = person.startsWith('p:') ? (await coach.prospects(owner)).some(p => 'p:' + p.id === person)
    : await accounts.isDownlineOf(owner, person, 1);
  if (!ok) return { error: 'Not on your board.' };
  const tag = TAGS.includes(String(body.tag || '').toLowerCase()) ? String(body.tag || '').toLowerCase() : '';
  let followUp = null;
  if (body.followUp) { const t = /^\d{4}-\d{2}-\d{2}$/.test(body.followUp) ? Date.parse(body.followUp + 'T12:00:00') : Number(body.followUp); if (t && !isNaN(t)) followUp = t; }
  const note = String(body.note || '').trim().slice(0, 1000);
  const row = await impl().put(owner, person, { note, followUp, tag });
  return { ok: true, note: row };
}

module.exports = { init, STAGES, TAGS, visible, board, save };