MySQL data layer: accounts, sessions, campaigns, burns (Marty: real concurrency)

Coolify MySQL (instantadpay-db) via DATABASE_URL; db.js bootstraps schema
and one-time imports the volume JSON. accounts/auth/ads are dual-mode: the
MySQL path uses guarded UPDATEs for the concurrent ad-serving hot path;
without DATABASE_URL the JSON stores remain (local dev). All data functions
async; server boots through db.init. Chain index stays a file: it is a
rebuildable cache of the blockchain, which remains the money truth.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

db.js

@@ -0,0 +1,118 @@
+// 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 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) };