instantadpay/messages.js

// Sponsor → downline messages: team comms delivered to members' on-site inbox
// (and by email). Distinct from solo ADS (which are paid and earn credits).
// Dual-mode: MySQL when DATABASE_URL is set, else a JSON file in the volume.
const fs = require('fs');
const path = require('path');
const db = require('./db');

let DATA_DIR = null;
const J = {
  db: null,
  FILE: () => path.join(DATA_DIR, 'sponsor-messages.json'),
  load() { try { this.db = JSON.parse(fs.readFileSync(this.FILE(), 'utf8')); } catch (e) { this.db = { nextId: 1, items: [] }; } },
  save() { try { fs.writeFileSync(this.FILE(), JSON.stringify(this.db)); } catch (e) {} }
};
function init(opts) { DATA_DIR = opts.dataDir; }
const DAY = 86400000;

// has this sponsor already broadcast within the last 24h? (1/day cap)
async function lastBroadcastAt(fromEmail) {
  const e = String(fromEmail || '').toLowerCase();
  if (db.enabled()) {
    const r = await db.q("SELECT MAX(sent) m FROM sponsor_messages WHERE from_email=? AND kind='broadcast'", [e]);
    return r.length && r[0].m ? Number(r[0].m) : 0;
  }
  if (!J.db) J.load();
  const mine = J.db.items.filter(i => i.fromEmail === e && (i.kind || 'broadcast') === 'broadcast').map(i => i.sent);
  return mine.length ? Math.max(...mine) : 0;
}
// deliver one message to many recipients (already-resolved emails). Returns count.
//
// `kind` separates the two things that used to share this channel:
//   'broadcast' — a person wrote it to their team. Rare, and worth interrupting for.
//   'notice'    — the system generated it (payout landed, purchase confirmed, and so on).
//                 These arrive constantly, so they belong in the inbox and must never pop
//                 a modal. Before this split they did, which meant a member with a backlog
//                 of payout notices got a fresh popup after every single ad view.
async function deliver(fromMember, fromEmail, recipients, subject, body, kind) {
  const k = kind === 'notice' ? 'notice' : 'broadcast';
  const now = Date.now();
  let n = 0;
  if (db.enabled()) {
    for (const to of recipients) {
      await db.q("INSERT INTO sponsor_messages (from_member,from_email,to_email,subject,body,sent,kind) VALUES (?,?,?,?,?,?,?)",
        [fromMember || 0, fromEmail, to, subject, body, now, k]);
      n++;
    }
  } else {
    if (!J.db) J.load();
    for (const to of recipients) {
      J.db.items.push({ id: J.db.nextId++, kind: k, fromMember: fromMember || 0, fromEmail, toEmail: to, subject, body, sent: now, readTs: 0 });
      n++;
    }
    J.save();
  }
  return n;
}
// Both kinds are inbox mail; only 'chat' is the separate two-way thread.
const isBroadcast = i => (i.kind || 'broadcast') !== 'chat';
// ...but only a message a PERSON wrote may interrupt with the modal.
const isHuman = i => (i.kind || 'broadcast') === 'broadcast';
const isChat = i => i.kind === 'chat';
async function inbox(email) {
  const e = String(email || '').toLowerCase();
  if (db.enabled()) {
    const rows = await db.q(`SELECT id, from_member, subject, body, sent, read_ts FROM sponsor_messages
      WHERE to_email=? AND kind<>'chat' ORDER BY sent DESC LIMIT 100`, [e]);
    return rows.map(r => ({ id: r.id, fromMember: r.from_member, subject: r.subject, body: r.body,
      sent: Number(r.sent), read: !!r.read_ts }));
  }
  if (!J.db) J.load();
  return J.db.items.filter(i => i.toEmail === e && isBroadcast(i)).sort((a, b) => b.sent - a.sent).slice(0, 100)
    .map(i => ({ id: i.id, fromMember: i.fromMember, subject: i.subject, body: i.body, sent: i.sent, read: !!i.readTs }));
}

// ── two-way sponsor CHAT (kind='chat'), plain-text 1:1 threads ──
async function sendChat(fromMember, fromEmail, toEmail, body) {
  const now = Date.now();
  const from = String(fromEmail || '').toLowerCase(), to = String(toEmail || '').toLowerCase();
  if (db.enabled()) {
    const r = await db.q("INSERT INTO sponsor_messages (from_member,from_email,to_email,subject,body,sent,kind) VALUES (?,?,?,'',?,?,'chat')",
      [fromMember || 0, from, to, body, now]);
    return { id: r.insertId, sent: now };
  }
  if (!J.db) J.load();
  const id = J.db.nextId++;
  J.db.items.push({ id, kind: 'chat', fromMember: fromMember || 0, fromEmail: from, toEmail: to, subject: '', body, sent: now, readTs: 0 });
  J.save();
  return { id, sent: now };
}
// messages between two members (both directions), id > afterId, oldest→newest
// last message from one member to another, any kind (0 if never)
async function lastFrom(fromEmail, toEmail) {
  const f = String(fromEmail || '').toLowerCase(), t = String(toEmail || '').toLowerCase();
  if (db.enabled()) { const r = await db.q('SELECT MAX(sent) s FROM sponsor_messages WHERE from_email=? AND to_email=?', [f, t]); return Number((r[0] && r[0].s) || 0); }
  if (!J.db) J.load();
  return J.db.items.filter(i => i.fromEmail === f && i.toEmail === t).reduce((m, i) => Math.max(m, i.sent || 0), 0);
}
async function thread(aEmail, bEmail, afterId = 0, limit = 300) {
  const a = String(aEmail || '').toLowerCase(), b = String(bEmail || '').toLowerCase();
  if (db.enabled()) {
    const rows = await db.q(`SELECT id, from_member, from_email, body, sent, read_ts FROM sponsor_messages
      WHERE kind='chat' AND id>? AND ((from_email=? AND to_email=?) OR (from_email=? AND to_email=?))
      ORDER BY id ASC LIMIT ?`, [Number(afterId) || 0, a, b, b, a, limit]);
    return rows.map(r => ({ id: r.id, fromMember: r.from_member, fromEmail: r.from_email, body: r.body, sent: Number(r.sent), read: !!r.read_ts }));
  }
  if (!J.db) J.load();
  return J.db.items.filter(i => isChat(i) && i.id > (Number(afterId) || 0)
      && ((i.fromEmail === a && i.toEmail === b) || (i.fromEmail === b && i.toEmail === a)))
    .sort((x, y) => x.id - y.id).slice(0, limit)
    .map(i => ({ id: i.id, fromMember: i.fromMember, fromEmail: i.fromEmail, body: i.body, sent: i.sent, read: !!i.readTs }));
}
// mark every chat message FROM other → email as read
async function markChatRead(email, otherEmail) {
  const e = String(email || '').toLowerCase(), o = String(otherEmail || '').toLowerCase();
  if (db.enabled()) {
    await db.q("UPDATE sponsor_messages SET read_ts=? WHERE kind='chat' AND to_email=? AND from_email=? AND read_ts IS NULL", [Date.now(), e, o]);
    return { ok: true };
  }
  if (!J.db) J.load();
  let touched = false;
  for (const i of J.db.items) if (isChat(i) && i.toEmail === e && i.fromEmail === o && !i.readTs) { i.readTs = Date.now(); touched = true; }
  if (touched) J.save();
  return { ok: true };
}
async function chatUnread(email) {
  const e = String(email || '').toLowerCase();
  if (db.enabled()) {
    const r = await db.q("SELECT COUNT(*) n FROM sponsor_messages WHERE kind='chat' AND to_email=? AND read_ts IS NULL", [e]);
    return r[0].n;
  }
  if (!J.db) J.load();
  return J.db.items.filter(i => isChat(i) && i.toEmail === e && !i.readTs).length;
}
// one row per conversation partner: newest message + my unread count from them
async function threadList(email) {
  const e = String(email || '').toLowerCase();
  let rows;
  if (db.enabled()) {
    rows = (await db.q(`SELECT from_email, to_email, body, sent, read_ts FROM sponsor_messages
      WHERE kind='chat' AND (from_email=? OR to_email=?) ORDER BY sent DESC LIMIT 800`, [e, e]))
      .map(r => ({ fromEmail: r.from_email, toEmail: r.to_email, body: r.body, sent: Number(r.sent), read: !!r.read_ts }));
  } else {
    if (!J.db) J.load();
    rows = J.db.items.filter(i => isChat(i) && (i.fromEmail === e || i.toEmail === e))
      .sort((a, b) => b.sent - a.sent)
      .map(i => ({ fromEmail: i.fromEmail, toEmail: i.toEmail, body: i.body, sent: i.sent, read: !!i.readTs }));
  }
  const byOther = new Map();
  for (const r of rows) {
    const other = r.fromEmail === e ? r.toEmail : r.fromEmail;
    if (!byOther.has(other)) byOther.set(other, { email: other, last: { body: r.body, sent: r.sent, fromMe: r.fromEmail === e }, unread: 0 });
    if (r.toEmail === e && !r.read) byOther.get(other).unread++;
  }
  return [...byOther.values()];
}
async function unreadCount(email) {
  const e = String(email || '').toLowerCase();
  if (db.enabled()) {
    const r = await db.q("SELECT COUNT(*) n FROM sponsor_messages WHERE to_email=? AND kind<>'chat' AND read_ts IS NULL", [e]);
    return r[0].n;
  }
  if (!J.db) J.load();
  return J.db.items.filter(i => i.toEmail === e && isBroadcast(i) && !i.readTs).length;
}
// the newest unread BROADCAST (for the sign-in modal; chat never pops the modal)
async function newestUnread(email) {
  const e = String(email || '').toLowerCase();
  if (db.enabled()) {
    const rows = await db.q(`SELECT id, from_member, subject, body, sent FROM sponsor_messages
      WHERE to_email=? AND kind='broadcast' AND read_ts IS NULL ORDER BY sent DESC LIMIT 1`, [e]);
    if (!rows.length) return null;
    const r = rows[0];
    return { id: r.id, fromMember: r.from_member, subject: r.subject, body: r.body, sent: Number(r.sent) };
  }
  if (!J.db) J.load();
  const u = J.db.items.filter(i => i.toEmail === e && isHuman(i) && !i.readTs).sort((a, b) => b.sent - a.sent)[0];
  return u ? { id: u.id, fromMember: u.fromMember, subject: u.subject, body: u.body, sent: u.sent } : null;
}
async function markRead(email, id) {
  const e = String(email || '').toLowerCase();
  if (db.enabled()) {
    await db.q('UPDATE sponsor_messages SET read_ts=? WHERE id=? AND to_email=? AND read_ts IS NULL', [Date.now(), Number(id), e]);
    return { ok: true };
  }
  if (!J.db) J.load();
  const i = J.db.items.find(x => x.id === Number(id) && x.toEmail === e);
  if (i && !i.readTs) { i.readTs = Date.now(); J.save(); }
  return { ok: true };
}

module.exports = { init, lastFrom, lastBroadcastAt, deliver, inbox, unreadCount, newestUnread, markRead,
  sendChat, thread, markChatRead, chatUnread, threadList };