instantadpay/db.js

// MySQL data layer (Marty: real concurrency over flat files).
// DATABASE_URL present  -> mysql2 pool, schema bootstrap, one-time import of
//                          any existing volume JSON (accounts/sessions/
//                          campaigns) so nothing is lost on the switch.
// DATABASE_URL absent   -> db.enabled() is false and the modules fall back to
//                          their JSON stores (local dev keeps working).
// The chain index stays a file: it is a rebuildable cache of the blockchain.
const fs = require('fs');
const path = require('path');

let pool = null;
let DATA_DIR = null;

function enabled() { return !!process.env.DATABASE_URL; }

async function init(opts) {
  DATA_DIR = opts.dataDir;
  if (!enabled()) return false;
  const mysql = require('mysql2/promise');
  pool = mysql.createPool(process.env.DATABASE_URL + '?connectionLimit=10&charset=utf8mb4');
  await bootstrap();
  await importJsonOnce();
  console.log('db: MySQL connected, schema ready');
  return true;
}
const q = async (sql, params) => (await pool.query(sql, params))[0];

async function bootstrap() {
  await q(`CREATE TABLE IF NOT EXISTS accounts (
    email VARCHAR(190) PRIMARY KEY,
    pass VARCHAR(200) NULL,
    sponsor_ref VARCHAR(32) NOT NULL DEFAULT '',
    code VARCHAR(16) NOT NULL UNIQUE,
    address VARCHAR(64) NULL UNIQUE,
    created BIGINT NOT NULL,
    last_seen BIGINT NULL
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
  await q(`CREATE TABLE IF NOT EXISTS sessions (
    token VARCHAR(80) PRIMARY KEY,
    email VARCHAR(190) NULL,
    address VARCHAR(64) NULL,
    member_id INT NOT NULL DEFAULT 0,
    expires BIGINT NOT NULL,
    INDEX (expires)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
  await q(`CREATE TABLE IF NOT EXISTS campaigns (
    id INT AUTO_INCREMENT PRIMARY KEY,
    owner_email VARCHAR(190) NOT NULL,
    member_id INT NOT NULL,
    type VARCHAR(12) NOT NULL,
    name VARCHAR(80) NOT NULL,
    target_url VARCHAR(500) NOT NULL,
    image_url VARCHAR(500) NULL,
    title VARCHAR(80) NULL,
    body VARCHAR(200) NULL,
    budget INT NOT NULL,
    spent INT NOT NULL DEFAULT 0,
    accrued INT NOT NULL DEFAULT 0,
    imps INT NOT NULL DEFAULT 0,
    clicks INT NOT NULL DEFAULT 0,
    batch_imps INT NOT NULL DEFAULT 0,
    status VARCHAR(12) NOT NULL DEFAULT 'active',
    last_day_charged VARCHAR(10) NULL,
    created BIGINT NOT NULL,
    INDEX (owner_email), INDEX (type, status), INDEX (member_id)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
  await q(`CREATE TABLE IF NOT EXISTS earned_credits (
    email VARCHAR(190) PRIMARY KEY,
    balance INT NOT NULL DEFAULT 0,
    granted_welcome TINYINT NOT NULL DEFAULT 0,
    updated BIGINT NOT NULL
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
  // additive columns (MySQL 8 has no IF NOT EXISTS for columns)
  const alterSafe = async sql => {
    try { await q(sql); }
    catch (e) { if (!['ER_DUP_FIELDNAME', 'ER_DUP_KEYNAME'].includes(e.code)) throw e; }
  };
  await alterSafe('ALTER TABLE accounts ADD COLUMN username VARCHAR(30) NULL');
  await alterSafe('ALTER TABLE accounts ADD UNIQUE KEY uq_username (username)');
  await alterSafe('ALTER TABLE accounts ADD COLUMN member_id INT NULL');
  await alterSafe('ALTER TABLE accounts ADD KEY idx_member (member_id)');
  await q(`CREATE TABLE IF NOT EXISTS daily_views (
    email VARCHAR(190) NOT NULL,
    day CHAR(10) NOT NULL,
    views INT NOT NULL DEFAULT 0,
    claimed TINYINT NOT NULL DEFAULT 0,
    last_ts BIGINT NOT NULL DEFAULT 0,
    PRIMARY KEY (email, day)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
  await q(`CREATE TABLE IF NOT EXISTS burns (
    id VARCHAR(32) PRIMARY KEY,
    member_id INT NOT NULL,
    amount INT NOT NULL,
    ref VARCHAR(64) NOT NULL,
    ts BIGINT NOT NULL,
    burned_tx VARCHAR(80) NULL,
    burned_at BIGINT NULL,
    INDEX (burned_tx)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
}

// one-time import: only when the tables are empty and JSON files exist
async function importJsonOnce() {
  const [{ n }] = await q('SELECT COUNT(*) n FROM accounts');
  if (n > 0) return;
  try {
    const a = JSON.parse(fs.readFileSync(path.join(DATA_DIR, 'accounts.json'), 'utf8'));
    for (const acc of Object.values(a.byEmail || {})) {
      await q('INSERT IGNORE INTO accounts (email,pass,sponsor_ref,code,address,created) VALUES (?,?,?,?,?,?)',
        [acc.email, acc.pass || null, String(acc.sponsorRef || acc.sponsorId || ''), acc.code, acc.address || null, acc.created || Date.now()]);
    }
    console.log('db: imported', Object.keys(a.byEmail || {}).length, 'account(s) from JSON');
  } catch (e) {}
  try {
    const c = JSON.parse(fs.readFileSync(path.join(DATA_DIR, 'campaigns.json'), 'utf8'));
    for (const cp of c.campaigns || []) {
      await q(`INSERT IGNORE INTO campaigns (owner_email,member_id,type,name,target_url,image_url,title,body,
        budget,spent,accrued,imps,clicks,batch_imps,status,last_day_charged,created)
        VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
        [cp.owner, cp.memberId, cp.type, cp.name, cp.targetUrl, cp.imageUrl || null, cp.title || null, cp.body || null,
         cp.budget, cp.spent || 0, cp.accrued || 0, cp.imps || 0, cp.clicks || 0, cp.batchImps || 0,
         cp.status, cp.lastDayCharged || null, cp.created || Date.now()]);
    }
    for (const b of c.burnsPending || []) {
      await q('INSERT IGNORE INTO burns (id,member_id,amount,ref,ts,burned_tx,burned_at) VALUES (?,?,?,?,?,?,?)',
        [b.id, b.memberId, b.amount, b.ref, b.ts, b.burnedTx || null, b.burnedAt || null]);
    }
    console.log('db: imported', (c.campaigns || []).length, 'campaign(s) from JSON');
  } catch (e) {}
  try {
    const s = JSON.parse(fs.readFileSync(path.join(DATA_DIR, 'sessions.json'), 'utf8'));
    for (const [tok, sess] of Object.entries(s)) {
      if (sess.expires > Date.now()) {
        await q('INSERT IGNORE INTO sessions (token,email,address,member_id,expires) VALUES (?,?,?,?,?)',
          [tok, sess.email || null, sess.address || null, sess.memberId || 0, sess.expires]);
      }
    }
  } catch (e) {}
}

module.exports = { init, enabled, q: (...a) => q(...a) };