Claude Opus 5andClaude Opus 5
71856b15e9
The repository carried two different servers side by side - the early
`justrp` prototype with its Python auth API, and a snapshot of the real
one under `server-code/`. Only the second one is a server anyone should
start from, so the prototype is gone and the real one moved to the root.
Taken from the live box, so the UI is the finished version (the tokens,
the county-records sheets and the variable fonts landed after the last
snapshot was pushed):
resources/[rp]/ rp_db, rp_core, rp_session, rp_loading, rp_ui,
rp_selftest, rp_dbtest
bin/ supervise.sh (keep-alive + console FIFO), rcon, init script
etc/schema.sql accounts, characters, transactions, inventory, vehicles
assets/fonts/ the bundled subsets
docs/SERVER.md how it is put together, resource by resource
install.sh turns a clean Ubuntu/Debian box into this server in one
command: recommended FXServer build, MariaDB with the schema, resources,
server.cfg + a generated database password, boot entry (systemd or
init.d), then it waits for the Cfx registration. --name/--port/--db-name
let a second server live on the same machine. Tested end to end on a
spare install root: build 25770 fetched, schema applied, boot entry
written, FXServer started and stopped exactly where a wrong licence key
should stop it.
No secrets travel with it: the licence key and the database password live
in data/secrets.cfg on the machine, and .gitignore now names it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
747 lines
26 KiB
JavaScript
747 lines
26 KiB
JavaScript
// ---------------------------------------------------------------------------
|
|
// rp_db - a MySQL/MariaDB client written for this server.
|
|
//
|
|
// Speaks the MySQL client/server protocol directly over a TCP socket: framing,
|
|
// handshake, mysql_native_password / caching_sha2_password (fast path), and
|
|
// the text result-set protocol. No external packages - only node built-ins.
|
|
//
|
|
// Deliberate choices:
|
|
// * CLIENT_MULTI_STATEMENTS is NOT negotiated, so a stacked-query injection
|
|
// ("'; DROP TABLE ..") is rejected by the server even if escaping failed.
|
|
// * CLIENT_LOCAL_FILES is NOT negotiated, so the server can never ask us to
|
|
// read a local file.
|
|
// * sql_mode is pinned at connect time, which also guarantees backslash
|
|
// escaping semantics that format() relies on.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const net = require('net');
|
|
const crypto = require('crypto');
|
|
|
|
// -- capability flags --------------------------------------------------------
|
|
const CAP = {
|
|
LONG_PASSWORD: 0x00000001,
|
|
FOUND_ROWS: 0x00000002,
|
|
LONG_FLAG: 0x00000004,
|
|
CONNECT_WITH_DB: 0x00000008,
|
|
LOCAL_FILES: 0x00000080,
|
|
PROTOCOL_41: 0x00000200,
|
|
TRANSACTIONS: 0x00002000,
|
|
SECURE_CONNECTION: 0x00008000,
|
|
MULTI_STATEMENTS: 0x00010000,
|
|
MULTI_RESULTS: 0x00020000,
|
|
PLUGIN_AUTH: 0x00080000,
|
|
CONNECT_ATTRS: 0x00100000,
|
|
PLUGIN_AUTH_LENENC: 0x00200000,
|
|
};
|
|
|
|
const CLIENT_CAPS =
|
|
CAP.LONG_PASSWORD | CAP.LONG_FLAG | CAP.CONNECT_WITH_DB | CAP.PROTOCOL_41 |
|
|
CAP.TRANSACTIONS | CAP.SECURE_CONNECTION | CAP.PLUGIN_AUTH | CAP.PLUGIN_AUTH_LENENC;
|
|
|
|
const COM = { QUIT: 0x01, QUERY: 0x03, PING: 0x0e };
|
|
|
|
// MySQL column type codes we care about when coercing text-protocol values.
|
|
const T_INT = new Set([1, 2, 3, 8, 9, 13]); // TINY SHORT LONG LONGLONG INT24 YEAR
|
|
const T_FLOAT = new Set([0, 4, 5, 246]); // DECIMAL FLOAT DOUBLE NEWDECIMAL
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Buffer helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
class Reader {
|
|
constructor(buf) { this.b = buf; this.o = 0; }
|
|
u8() { return this.b[this.o++]; }
|
|
u16() { const v = this.b.readUInt16LE(this.o); this.o += 2; return v; }
|
|
u32() { const v = this.b.readUInt32LE(this.o); this.o += 4; return v; }
|
|
skip(n) { this.o += n; return this; }
|
|
bytes(n) { const v = this.b.subarray(this.o, this.o + n); this.o += n; return v; }
|
|
nulStr() {
|
|
let i = this.b.indexOf(0, this.o);
|
|
if (i < 0) i = this.b.length;
|
|
const s = this.b.toString('utf8', this.o, i);
|
|
this.o = i + 1;
|
|
return s;
|
|
}
|
|
// length-encoded integer; returns null for the 0xfb NULL marker
|
|
lenenc() {
|
|
const f = this.b[this.o++];
|
|
if (f < 0xfb) return f;
|
|
if (f === 0xfb) return null;
|
|
if (f === 0xfc) { const v = this.b.readUInt16LE(this.o); this.o += 2; return v; }
|
|
if (f === 0xfd) { const v = this.b.readUIntLE(this.o, 3); this.o += 3; return v; }
|
|
const v = this.b.readBigUInt64LE(this.o); this.o += 8;
|
|
return Number(v);
|
|
}
|
|
lenencStr() {
|
|
const n = this.lenenc();
|
|
if (n === null) return null;
|
|
return this.bytes(n).toString('utf8');
|
|
}
|
|
remaining() { return this.b.length - this.o; }
|
|
}
|
|
|
|
class Writer {
|
|
constructor() { this.parts = []; }
|
|
u8(v) { const b = Buffer.alloc(1); b[0] = v; this.parts.push(b); return this; }
|
|
u16(v) { const b = Buffer.alloc(2); b.writeUInt16LE(v); this.parts.push(b); return this; }
|
|
u32(v) { const b = Buffer.alloc(4); b.writeUInt32LE(v >>> 0); this.parts.push(b); return this; }
|
|
zeros(n) { this.parts.push(Buffer.alloc(n)); return this; }
|
|
raw(b) { this.parts.push(b); return this; }
|
|
nulStr(s) { this.parts.push(Buffer.from(s, 'utf8')); return this.u8(0); }
|
|
lenencBuf(b) {
|
|
if (b.length < 0xfb) this.u8(b.length);
|
|
else if (b.length < 0x10000) { this.u8(0xfc).u16(b.length); }
|
|
else { const t = Buffer.alloc(4); t.writeUIntLE(b.length, 0, 3); this.u8(0xfd).raw(t.subarray(0, 3)); }
|
|
return this.raw(b);
|
|
}
|
|
done() { return Buffer.concat(this.parts); }
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Authentication
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const sha1 = (d) => crypto.createHash('sha1').update(d).digest();
|
|
const sha256 = (d) => crypto.createHash('sha256').update(d).digest();
|
|
|
|
function xorBuf(a, b) {
|
|
const out = Buffer.alloc(a.length);
|
|
for (let i = 0; i < a.length; i++) out[i] = a[i] ^ b[i % b.length];
|
|
return out;
|
|
}
|
|
|
|
// SHA1(pw) XOR SHA1( scramble + SHA1(SHA1(pw)) )
|
|
function authNative(password, scramble) {
|
|
if (!password) return Buffer.alloc(0);
|
|
const s1 = sha1(Buffer.from(password, 'utf8'));
|
|
const s2 = sha1(s1);
|
|
return xorBuf(s1, sha1(Buffer.concat([scramble, s2])));
|
|
}
|
|
|
|
// SHA256(pw) XOR SHA256( SHA256(SHA256(pw)) + scramble ) [fast path only]
|
|
function authCachingSha2(password, scramble) {
|
|
if (!password) return Buffer.alloc(0);
|
|
const s1 = sha256(Buffer.from(password, 'utf8'));
|
|
const s2 = sha256(s1);
|
|
return xorBuf(s1, sha256(Buffer.concat([s2, scramble])));
|
|
}
|
|
|
|
function authResponse(plugin, password, scramble) {
|
|
switch (plugin) {
|
|
case 'mysql_native_password': return authNative(password, scramble);
|
|
case 'caching_sha2_password': return authCachingSha2(password, scramble);
|
|
case 'mysql_clear_password': return Buffer.concat([Buffer.from(password, 'utf8'), Buffer.alloc(1)]);
|
|
default: return null; // unsupported -> connection fails loudly
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Value escaping. Every query goes through format(); values are never
|
|
// concatenated into SQL anywhere else in this codebase.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const ESCAPE_MAP = {
|
|
'\0': '\\0', '\b': '\\b', '\t': '\\t', '\n': '\\n',
|
|
'\r': '\\r', '\x1a': '\\Z', '"': '\\"', "'": "\\'", '\\': '\\\\',
|
|
};
|
|
|
|
function escapeString(s) {
|
|
return "'" + s.replace(/[\0\b\t\n\r\x1a"'\\]/g, (c) => ESCAPE_MAP[c]) + "'";
|
|
}
|
|
|
|
function pad2(n) { return n < 10 ? '0' + n : '' + n; }
|
|
|
|
function escapeValue(v) {
|
|
if (v === null || v === undefined) return 'NULL';
|
|
switch (typeof v) {
|
|
case 'boolean': return v ? '1' : '0';
|
|
case 'number':
|
|
if (!Number.isFinite(v)) throw new Error('cannot bind non-finite number');
|
|
return String(v);
|
|
case 'bigint': return String(v);
|
|
case 'string': return escapeString(v);
|
|
case 'object':
|
|
if (v instanceof Date) {
|
|
return escapeString(
|
|
`${v.getFullYear()}-${pad2(v.getMonth() + 1)}-${pad2(v.getDate())} ` +
|
|
`${pad2(v.getHours())}:${pad2(v.getMinutes())}:${pad2(v.getSeconds())}`
|
|
);
|
|
}
|
|
if (Buffer.isBuffer(v)) return 'x' + escapeString(v.toString('hex'));
|
|
if (Array.isArray(v)) return v.map(escapeValue).join(', ');
|
|
return escapeString(JSON.stringify(v)); // plain objects -> JSON columns
|
|
default:
|
|
throw new Error('cannot bind value of type ' + typeof v);
|
|
}
|
|
}
|
|
|
|
// Replace `?` placeholders, skipping any that appear inside string/identifier
|
|
// literals or comments so that user data containing '?' can never shift binding.
|
|
function format(sql, params) {
|
|
if (!params || (Array.isArray(params) && params.length === 0)) return sql;
|
|
const list = Array.isArray(params) ? params : [params];
|
|
let out = '';
|
|
let pi = 0;
|
|
let i = 0;
|
|
const n = sql.length;
|
|
|
|
while (i < n) {
|
|
const c = sql[i];
|
|
|
|
if (c === "'" || c === '"' || c === '`') {
|
|
const quote = c;
|
|
out += c; i++;
|
|
while (i < n) {
|
|
if (sql[i] === '\\' && quote !== '`') { out += sql[i] + (sql[i + 1] || ''); i += 2; continue; }
|
|
if (sql[i] === quote) {
|
|
if (sql[i + 1] === quote) { out += quote + quote; i += 2; continue; } // doubled quote
|
|
out += quote; i++; break;
|
|
}
|
|
out += sql[i]; i++;
|
|
}
|
|
continue;
|
|
}
|
|
if (c === '-' && sql[i + 1] === '-') { while (i < n && sql[i] !== '\n') out += sql[i++]; continue; }
|
|
if (c === '#') { while (i < n && sql[i] !== '\n') out += sql[i++]; continue; }
|
|
if (c === '/' && sql[i + 1] === '*') {
|
|
out += '/*'; i += 2;
|
|
while (i < n && !(sql[i] === '*' && sql[i + 1] === '/')) out += sql[i++];
|
|
out += '*/'; i += 2; continue;
|
|
}
|
|
if (c === '?') {
|
|
if (pi >= list.length) throw new Error('not enough parameters for query');
|
|
out += escapeValue(list[pi++]);
|
|
i++; continue;
|
|
}
|
|
out += c; i++;
|
|
}
|
|
|
|
if (pi !== list.length) throw new Error(`parameter count mismatch (query takes ${pi}, got ${list.length})`);
|
|
return out;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Connection
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const MAX_PAYLOAD = 0xffffff;
|
|
|
|
class Connection {
|
|
constructor(opts, id) {
|
|
this.opts = opts;
|
|
this.id = id;
|
|
this.socket = null;
|
|
this.buf = Buffer.alloc(0);
|
|
this.seq = 0;
|
|
this.state = 'offline';
|
|
this.busy = false;
|
|
this.pinned = false; // reserved by a transaction; pump() must skip it
|
|
this.pending = null; // {resolve, reject, sql}
|
|
this.rs = null; // in-flight result set
|
|
this.multi = null; // reassembly buffer for >16MB payloads
|
|
this.serverCaps = 0;
|
|
this.onIdle = null;
|
|
this.retryDelay = 1000;
|
|
}
|
|
|
|
log(msg) { console.log(`^5[rp_db]^7 conn#${this.id} ${msg}`); }
|
|
err(msg) { console.log(`^1[rp_db]^7 conn#${this.id} ${msg}`); }
|
|
|
|
connect() {
|
|
if (this.state !== 'offline') return;
|
|
this.state = 'connecting';
|
|
this.buf = Buffer.alloc(0);
|
|
this.seq = 0;
|
|
|
|
const sock = net.createConnection({ host: this.opts.host, port: this.opts.port });
|
|
this.socket = sock;
|
|
sock.setNoDelay(true);
|
|
|
|
sock.on('data', (d) => {
|
|
this.buf = this.buf.length ? Buffer.concat([this.buf, d]) : d;
|
|
this.drainPackets();
|
|
});
|
|
sock.on('error', (e) => this.fail(e.message));
|
|
sock.on('close', () => { if (this.state !== 'offline') this.fail('connection closed'); });
|
|
}
|
|
|
|
fail(reason) {
|
|
const wasReady = this.state === 'ready' || this.state === 'query';
|
|
this.state = 'offline';
|
|
if (this.socket) { this.socket.destroy(); this.socket = null; }
|
|
if (this.pending) {
|
|
const p = this.pending; this.pending = null; this.busy = false;
|
|
p.reject(new Error(reason));
|
|
}
|
|
this.busy = false;
|
|
this.pinned = false; // a dropped connection must not stay reserved
|
|
this.rs = null;
|
|
if (wasReady) this.err(`lost connection: ${reason}`);
|
|
else this.err(`connect failed: ${reason}`);
|
|
// exponential backoff, capped
|
|
setTimeout(() => this.connect(), this.retryDelay);
|
|
this.retryDelay = Math.min(this.retryDelay * 2, 30000);
|
|
}
|
|
|
|
// Split the stream into protocol packets, reassembling 16MB-split payloads.
|
|
drainPackets() {
|
|
for (;;) {
|
|
if (this.buf.length < 4) return;
|
|
const len = this.buf.readUIntLE(0, 3);
|
|
if (this.buf.length < 4 + len) return;
|
|
const seq = this.buf[3];
|
|
const payload = this.buf.subarray(4, 4 + len);
|
|
this.buf = this.buf.subarray(4 + len);
|
|
this.seq = (seq + 1) & 0xff;
|
|
|
|
if (len === MAX_PAYLOAD) { // more to come
|
|
this.multi = this.multi ? Buffer.concat([this.multi, payload]) : payload;
|
|
continue;
|
|
}
|
|
let full = payload;
|
|
if (this.multi) { full = Buffer.concat([this.multi, payload]); this.multi = null; }
|
|
|
|
try { this.onPacket(full); }
|
|
catch (e) { this.err(`packet handling error: ${e.message}`); this.fail(e.message); return; }
|
|
}
|
|
}
|
|
|
|
send(payload) {
|
|
if (!this.socket) return;
|
|
const chunks = [];
|
|
let off = 0;
|
|
do {
|
|
const slice = payload.subarray(off, off + MAX_PAYLOAD);
|
|
const head = Buffer.alloc(4);
|
|
head.writeUIntLE(slice.length, 0, 3);
|
|
head[3] = this.seq & 0xff;
|
|
this.seq = (this.seq + 1) & 0xff;
|
|
chunks.push(head, slice);
|
|
off += slice.length;
|
|
} while (off < payload.length);
|
|
this.socket.write(Buffer.concat(chunks));
|
|
}
|
|
|
|
sendCommand(cmd, text) {
|
|
this.seq = 0; // every command starts a new sequence
|
|
const body = text ? Buffer.from(text, 'utf8') : Buffer.alloc(0);
|
|
this.send(Buffer.concat([Buffer.from([cmd]), body]));
|
|
}
|
|
|
|
onPacket(p) {
|
|
switch (this.state) {
|
|
case 'connecting': return this.onHandshake(p);
|
|
case 'auth': return this.onAuthResult(p);
|
|
case 'init': return this.onInitResult(p);
|
|
case 'query': return this.onQueryPacket(p);
|
|
default:
|
|
this.err(`unexpected packet in state ${this.state}`);
|
|
}
|
|
}
|
|
|
|
// -- handshake -------------------------------------------------------------
|
|
onHandshake(p) {
|
|
const r = new Reader(p);
|
|
const protocol = r.u8();
|
|
if (protocol === 0xff) return this.fail(this.readError(p).message);
|
|
if (protocol !== 10) return this.fail(`unsupported protocol version ${protocol}`);
|
|
|
|
this.serverVersion = r.nulStr();
|
|
r.u32(); // connection id
|
|
const scramble1 = r.bytes(8);
|
|
r.skip(1); // filler
|
|
let caps = r.u16();
|
|
let scramble2 = Buffer.alloc(0);
|
|
let plugin = 'mysql_native_password';
|
|
|
|
if (r.remaining() > 0) {
|
|
r.u8(); // charset
|
|
r.u16(); // status flags
|
|
caps |= r.u16() << 16;
|
|
const authDataLen = r.u8();
|
|
r.skip(10);
|
|
if (caps & CAP.SECURE_CONNECTION) {
|
|
const n = Math.max(13, authDataLen - 8);
|
|
scramble2 = r.bytes(n);
|
|
// trailing NUL is part of the padding, not the scramble
|
|
if (scramble2.length && scramble2[scramble2.length - 1] === 0) {
|
|
scramble2 = scramble2.subarray(0, scramble2.length - 1);
|
|
}
|
|
}
|
|
if (caps & CAP.PLUGIN_AUTH) plugin = r.nulStr();
|
|
}
|
|
|
|
this.serverCaps = caps >>> 0;
|
|
this.scramble = Buffer.concat([scramble1, scramble2]);
|
|
|
|
const resp = authResponse(plugin, this.opts.password, this.scramble);
|
|
if (resp === null) return this.fail(`server requested unsupported auth plugin '${plugin}'`);
|
|
|
|
const w = new Writer();
|
|
w.u32(CLIENT_CAPS).u32(MAX_PAYLOAD).u8(45 /* utf8mb4_general_ci */).zeros(23);
|
|
w.nulStr(this.opts.user);
|
|
w.lenencBuf(resp);
|
|
w.nulStr(this.opts.database);
|
|
w.nulStr(plugin);
|
|
|
|
this.authPlugin = plugin;
|
|
this.state = 'auth';
|
|
this.send(w.done());
|
|
}
|
|
|
|
onAuthResult(p) {
|
|
const marker = p[0];
|
|
|
|
if (marker === 0x00) return this.afterAuth();
|
|
if (marker === 0xff) return this.fail(this.readError(p).message);
|
|
|
|
if (marker === 0xfe) { // AuthSwitchRequest
|
|
const r = new Reader(p); r.u8();
|
|
const plugin = r.nulStr();
|
|
let scramble = r.bytes(r.remaining());
|
|
if (scramble.length && scramble[scramble.length - 1] === 0) scramble = scramble.subarray(0, scramble.length - 1);
|
|
const resp = authResponse(plugin, this.opts.password, scramble);
|
|
if (resp === null) return this.fail(`server switched to unsupported auth plugin '${plugin}'`);
|
|
this.scramble = scramble;
|
|
this.authPlugin = plugin;
|
|
return this.send(resp);
|
|
}
|
|
|
|
if (marker === 0x01) { // caching_sha2 extra auth data
|
|
const status = p[1];
|
|
if (status === 0x03) return; // fast auth ok -> OK packet follows
|
|
if (status === 0x04) {
|
|
return this.fail(
|
|
'caching_sha2_password requires a full authentication exchange (TLS or the ' +
|
|
'server public key), which this driver does not implement. Give the database ' +
|
|
"user mysql_native_password, e.g. ALTER USER .. IDENTIFIED VIA mysql_native_password."
|
|
);
|
|
}
|
|
return this.fail(`unexpected auth continuation 0x${status.toString(16)}`);
|
|
}
|
|
|
|
this.fail(`unexpected auth response 0x${marker.toString(16)}`);
|
|
}
|
|
|
|
// Pin session settings, then announce readiness.
|
|
afterAuth() {
|
|
this.retryDelay = 1000;
|
|
this.state = 'init';
|
|
this.sendCommand(
|
|
COM.QUERY,
|
|
"SET NAMES utf8mb4, sql_mode='STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION', " +
|
|
"time_zone='+00:00', autocommit=1"
|
|
);
|
|
}
|
|
|
|
onInitResult(p) {
|
|
if (p[0] === 0xff) return this.fail(this.readError(p).message);
|
|
this.state = 'ready';
|
|
this.busy = false;
|
|
this.log(`ready (server ${this.serverVersion}, auth ${this.authPlugin})`);
|
|
if (this.onIdle) this.onIdle(this);
|
|
}
|
|
|
|
// -- errors ----------------------------------------------------------------
|
|
readError(p) {
|
|
const r = new Reader(p);
|
|
r.u8();
|
|
const code = r.u16();
|
|
let sqlState = '';
|
|
if (p[3] === 0x23) { r.skip(1); sqlState = r.bytes(5).toString('ascii'); }
|
|
const message = p.subarray(r.o).toString('utf8');
|
|
const e = new Error(`${message} (errno ${code}${sqlState ? ', sqlstate ' + sqlState : ''})`);
|
|
e.errno = code; e.sqlState = sqlState;
|
|
return e;
|
|
}
|
|
|
|
readOk(p) {
|
|
const r = new Reader(p);
|
|
r.u8();
|
|
const affectedRows = r.lenenc() || 0;
|
|
const insertId = r.lenenc() || 0;
|
|
return { affectedRows, insertId };
|
|
}
|
|
|
|
isEof(p) { return p[0] === 0xfe && p.length < 9; }
|
|
|
|
// -- queries ---------------------------------------------------------------
|
|
query(sql) {
|
|
return new Promise((resolve, reject) => {
|
|
if (this.state !== 'ready') return reject(new Error('connection not ready'));
|
|
this.busy = true;
|
|
this.state = 'query';
|
|
this.pending = { resolve, reject, sql };
|
|
this.rs = { stage: 'head', columns: [], rows: [], colCount: 0 };
|
|
this.sendCommand(COM.QUERY, sql);
|
|
});
|
|
}
|
|
|
|
finish(value, error) {
|
|
const p = this.pending;
|
|
this.pending = null;
|
|
this.rs = null;
|
|
this.state = 'ready';
|
|
this.busy = false;
|
|
if (p) { error ? p.reject(error) : p.resolve(value); }
|
|
if (this.onIdle) this.onIdle(this);
|
|
}
|
|
|
|
onQueryPacket(p) {
|
|
const rs = this.rs;
|
|
|
|
if (rs.stage === 'head') {
|
|
if (p[0] === 0x00) return this.finish(this.readOk(p));
|
|
if (p[0] === 0xff) return this.finish(null, this.readError(p));
|
|
if (p[0] === 0xfb) return this.finish(null, new Error('LOCAL INFILE requested but not enabled'));
|
|
const r = new Reader(p);
|
|
rs.colCount = r.lenenc();
|
|
rs.stage = 'columns';
|
|
return;
|
|
}
|
|
|
|
if (rs.stage === 'columns') {
|
|
if (this.isEof(p)) { rs.stage = 'rows'; return; }
|
|
const r = new Reader(p);
|
|
r.lenencStr(); r.lenencStr(); r.lenencStr(); r.lenencStr(); // catalog, schema, table, org_table
|
|
const name = r.lenencStr();
|
|
r.lenencStr(); // org_name
|
|
r.lenenc(); // length of fixed fields
|
|
r.u16(); // charset
|
|
r.u32(); // column length
|
|
const type = r.u8();
|
|
rs.columns.push({ name, type });
|
|
return;
|
|
}
|
|
|
|
// stage === 'rows'
|
|
if (p[0] === 0xff) return this.finish(null, this.readError(p));
|
|
if (this.isEof(p)) return this.finish(rs.rows);
|
|
|
|
const r = new Reader(p);
|
|
const row = {};
|
|
for (let i = 0; i < rs.colCount; i++) {
|
|
const col = rs.columns[i];
|
|
const raw = r.lenencStr();
|
|
row[col.name] = raw === null ? null : coerce(raw, col.type);
|
|
}
|
|
rs.rows.push(row);
|
|
}
|
|
|
|
ping() {
|
|
if (this.state !== 'ready') return;
|
|
this.busy = true;
|
|
this.state = 'query';
|
|
this.pending = {
|
|
resolve: () => {}, reject: (e) => this.err(`ping failed: ${e.message}`), sql: 'PING',
|
|
};
|
|
this.rs = { stage: 'head', columns: [], rows: [], colCount: 0 };
|
|
this.sendCommand(COM.PING, '');
|
|
}
|
|
}
|
|
|
|
// Text protocol hands us strings; give Lua sensible types back.
|
|
function coerce(raw, type) {
|
|
if (T_INT.has(type)) {
|
|
const n = Number(raw);
|
|
return Number.isSafeInteger(n) ? n : raw; // keep huge BIGINTs exact as strings
|
|
}
|
|
if (T_FLOAT.has(type)) {
|
|
const n = Number(raw);
|
|
return Number.isNaN(n) ? raw : n;
|
|
}
|
|
return raw; // strings, dates, JSON, blobs
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Pool
|
|
// ---------------------------------------------------------------------------
|
|
|
|
class Pool {
|
|
constructor(opts) {
|
|
this.opts = opts;
|
|
this.queue = [];
|
|
this.conns = [];
|
|
for (let i = 0; i < opts.size; i++) {
|
|
const c = new Connection(opts, i + 1);
|
|
c.onIdle = () => this.pump();
|
|
this.conns.push(c);
|
|
}
|
|
}
|
|
|
|
start() {
|
|
this.conns.forEach((c, i) => setTimeout(() => c.connect(), i * 120));
|
|
setInterval(() => {
|
|
for (const c of this.conns) if (c.state === 'ready' && !c.busy) c.ping();
|
|
}, 30000);
|
|
}
|
|
|
|
get ready() { return this.conns.some((c) => c.state === 'ready'); }
|
|
|
|
free() { return this.conns.find((c) => c.state === 'ready' && !c.busy && !c.pinned); }
|
|
|
|
pump() {
|
|
while (this.queue.length) {
|
|
const conn = this.free();
|
|
if (!conn) return;
|
|
const job = this.queue.shift();
|
|
clearTimeout(job.timer);
|
|
if (job.settled) continue;
|
|
job.settled = true;
|
|
conn.query(job.sql).then(job.resolve, job.reject);
|
|
}
|
|
}
|
|
|
|
exec(sql) {
|
|
return new Promise((resolve, reject) => {
|
|
const job = { sql, resolve, reject, settled: false };
|
|
// Each job owns its own deadline, so a query still fails loudly when the
|
|
// database is down and no connection ever becomes free to trigger pump().
|
|
job.timer = setTimeout(() => {
|
|
if (job.settled) return;
|
|
job.settled = true;
|
|
const i = this.queue.indexOf(job);
|
|
if (i >= 0) this.queue.splice(i, 1);
|
|
reject(new Error('timed out waiting for a database connection'));
|
|
}, this.opts.queueTimeout);
|
|
this.queue.push(job);
|
|
this.pump();
|
|
});
|
|
}
|
|
|
|
// Pins one connection for the whole transaction so no other query can
|
|
// interleave between START TRANSACTION and COMMIT.
|
|
async transaction(statements) {
|
|
const held = await this.acquire();
|
|
try {
|
|
await held.query('START TRANSACTION');
|
|
for (const st of statements) await held.query(st);
|
|
await held.query('COMMIT');
|
|
return true;
|
|
} catch (e) {
|
|
try { await held.query('ROLLBACK'); } catch (_) { /* connection already gone */ }
|
|
throw e;
|
|
} finally {
|
|
held.release();
|
|
}
|
|
}
|
|
|
|
acquire() {
|
|
return new Promise((resolve, reject) => {
|
|
const started = Date.now();
|
|
const attempt = () => {
|
|
const conn = this.free();
|
|
if (conn) {
|
|
conn.pinned = true;
|
|
return resolve({
|
|
query: (sql) => conn.query(sql),
|
|
release: () => { conn.pinned = false; this.pump(); },
|
|
});
|
|
}
|
|
if (Date.now() - started > this.opts.queueTimeout) {
|
|
return reject(new Error('timed out waiting for a database connection'));
|
|
}
|
|
setTimeout(attempt, 25);
|
|
};
|
|
attempt();
|
|
});
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// FiveM bridge - exposes the pool to the Lua resources as exports.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// Lua cannot put nil inside a table without creating a hole, so callers pass
|
|
// this sentinel when they need a real SQL NULL. rp_db exposes it as DB.NULL.
|
|
const NULL_SENTINEL = 'RP_NULL';
|
|
|
|
function unsentinel(v) { return v === NULL_SENTINEL ? null : v; }
|
|
|
|
function normalizeParams(p) {
|
|
if (p === null || p === undefined) return [];
|
|
if (Array.isArray(p)) return p.map(unsentinel);
|
|
if (typeof p === 'object') {
|
|
const keys = Object.keys(p);
|
|
if (keys.length === 0) return []; // Lua {} arrives as an empty map
|
|
if (keys.every((k) => /^[0-9]+$/.test(k))) { // Lua array table
|
|
return keys.sort((a, b) => a - b).map((k) => unsentinel(p[k]));
|
|
}
|
|
return [p]; // a real object -> JSON column
|
|
}
|
|
return [unsentinel(p)];
|
|
}
|
|
|
|
const pool = new Pool({
|
|
host: GetConvar('rp_db_host', '127.0.0.1'),
|
|
port: parseInt(GetConvar('rp_db_port', '3306'), 10),
|
|
user: GetConvar('rp_db_user', 'rp'),
|
|
password: GetConvar('rp_db_pass', ''),
|
|
database: GetConvar('rp_db_name', 'rp'),
|
|
size: parseInt(GetConvar('rp_db_pool', '4'), 10),
|
|
queueTimeout: 20000,
|
|
});
|
|
|
|
// Run a query and hand the result to a Lua callback as (result, err).
|
|
// The SQL *template* is logged on failure, never the formatted text, so bound
|
|
// values (password hashes and the like) stay out of the console.
|
|
function run(sql, params, cb, shape) {
|
|
const done = (result, err) => {
|
|
if (err) {
|
|
console.log(`^1[rp_db]^7 query failed: ${err}`);
|
|
console.log(`^1[rp_db]^7 in: ${String(sql).trim().slice(0, 200)}`);
|
|
}
|
|
if (typeof cb === 'function') {
|
|
try { cb(result, err || null); }
|
|
catch (e) { console.log(`^1[rp_db]^7 callback threw: ${e.message}`); }
|
|
}
|
|
};
|
|
|
|
let text;
|
|
try { text = format(sql, normalizeParams(params)); }
|
|
catch (e) { return done(null, e.message); }
|
|
|
|
pool.exec(text).then(
|
|
(res) => { try { done(shape(res), null); } catch (e) { done(null, e.message); } },
|
|
(e) => done(null, e.message)
|
|
);
|
|
}
|
|
|
|
const rowsOf = (res) => (Array.isArray(res) ? res : []);
|
|
|
|
global.exports('query', (sql, params, cb) => run(sql, params, cb, (r) => r));
|
|
global.exports('single', (sql, params, cb) => run(sql, params, cb, (r) => rowsOf(r)[0] || null));
|
|
global.exports('scalar', (sql, params, cb) => run(sql, params, cb, (r) => {
|
|
const row = rowsOf(r)[0];
|
|
if (!row) return null;
|
|
const k = Object.keys(row)[0];
|
|
return k === undefined ? null : row[k];
|
|
}));
|
|
global.exports('insert', (sql, params, cb) => run(sql, params, cb, (r) => (r && r.insertId) || 0));
|
|
global.exports('update', (sql, params, cb) => run(sql, params, cb, (r) => (r && r.affectedRows) || 0));
|
|
|
|
// statements: array of {query = ..., values = {...}} tables from Lua
|
|
global.exports('transaction', (statements, cb) => {
|
|
let list;
|
|
try {
|
|
const arr = Array.isArray(statements) ? statements : Object.values(statements || {});
|
|
list = arr.map((s) => format(s.query, normalizeParams(s.values)));
|
|
} catch (e) {
|
|
console.log(`^1[rp_db]^7 transaction rejected: ${e.message}`);
|
|
if (typeof cb === 'function') cb(false, e.message);
|
|
return;
|
|
}
|
|
pool.transaction(list).then(
|
|
() => { if (typeof cb === 'function') cb(true, null); },
|
|
(e) => {
|
|
console.log(`^1[rp_db]^7 transaction rolled back: ${e.message}`);
|
|
if (typeof cb === 'function') cb(false, e.message);
|
|
}
|
|
);
|
|
});
|
|
|
|
global.exports('isReady', () => pool.ready);
|
|
global.exports('nullSentinel', () => NULL_SENTINEL);
|
|
|
|
pool.start();
|