Part 1 final build: Los Santos RP written on camera by the agent
rp_core (sessions, scrypt auth, self-written MySQL-wire driver rp_db), county-records NUI: residency-file auth, identity-record character intake, survey-grid spawn, procedural night-city loading screen. Secrets replaced with .example files. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -0,0 +1,15 @@
|
||||
fx_version 'cerulean'
|
||||
game 'gta5'
|
||||
|
||||
name 'rp_db'
|
||||
description 'MariaDB client speaking the MySQL wire protocol directly - no external packages'
|
||||
author 'Los Santos RP'
|
||||
version '1.0.0'
|
||||
|
||||
server_scripts {
|
||||
'server/mysql.js',
|
||||
}
|
||||
|
||||
files {
|
||||
'lib/db.lua',
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
-- ---------------------------------------------------------------------------
|
||||
-- DB - Lua front end for rp_db.
|
||||
--
|
||||
-- Loaded *into* the consuming resource with:
|
||||
-- server_scripts { '@rp_db/lib/db.lua', 'server/whatever.lua' }
|
||||
--
|
||||
-- so that Citizen.Await suspends the caller's own coroutine instead of trying
|
||||
-- to yield across a cross-resource export call.
|
||||
--
|
||||
-- Every function comes in two flavours:
|
||||
-- DB.Query(sql, params) -> rows, err (blocking, inside a thread)
|
||||
-- DB.QueryAsync(sql, params, cb) -> cb(rows, err) (callback, anywhere)
|
||||
--
|
||||
-- Values are always passed as `?` parameters. Never build SQL by concatenation.
|
||||
-- Use DB.NULL where you need a literal SQL NULL (Lua nil cannot survive a table).
|
||||
-- ---------------------------------------------------------------------------
|
||||
|
||||
DB = {}
|
||||
|
||||
DB.NULL = 'RP_NULL'
|
||||
|
||||
local backend = exports.rp_db
|
||||
|
||||
local function async(method)
|
||||
return function(sql, params, cb)
|
||||
backend[method](backend, sql, params or {}, cb)
|
||||
end
|
||||
end
|
||||
|
||||
local function blocking(method)
|
||||
return function(sql, params)
|
||||
local p = promise.new()
|
||||
backend[method](backend, sql, params or {}, function(result, err)
|
||||
-- wrapped: a promise resolved with nil would never wake the coroutine
|
||||
p:resolve({ result = result, err = err })
|
||||
end)
|
||||
local r = Citizen.Await(p)
|
||||
if r.err then return nil, r.err end
|
||||
return r.result
|
||||
end
|
||||
end
|
||||
|
||||
--- Returns all rows as an array of tables (empty table when nothing matched).
|
||||
DB.Query = blocking('query')
|
||||
--- Returns the first row as a table, or nil.
|
||||
DB.Single = blocking('single')
|
||||
--- Returns the first column of the first row, or nil.
|
||||
DB.Scalar = blocking('scalar')
|
||||
--- Returns the AUTO_INCREMENT id produced by an INSERT.
|
||||
DB.Insert = blocking('insert')
|
||||
--- Returns the number of rows changed by an UPDATE/DELETE.
|
||||
DB.Update = blocking('update')
|
||||
|
||||
DB.QueryAsync = async('query')
|
||||
DB.SingleAsync = async('single')
|
||||
DB.ScalarAsync = async('scalar')
|
||||
DB.InsertAsync = async('insert')
|
||||
DB.UpdateAsync = async('update')
|
||||
|
||||
--- All statements commit together or none of them do.
|
||||
--- statements: { { query = 'UPDATE ..', values = { 1, 2 } }, ... }
|
||||
function DB.Transaction(statements)
|
||||
local p = promise.new()
|
||||
backend:transaction(statements, function(ok, err)
|
||||
p:resolve({ ok = ok, err = err })
|
||||
end)
|
||||
local r = Citizen.Await(p)
|
||||
if not r.ok then return false, r.err end
|
||||
return true
|
||||
end
|
||||
|
||||
function DB.TransactionAsync(statements, cb)
|
||||
backend:transaction(statements, cb)
|
||||
end
|
||||
|
||||
--- Blocks until the connection pool has at least one live connection.
|
||||
--- Returns false if the database never came up within `timeout` ms.
|
||||
function DB.WaitReady(timeout)
|
||||
local waited, step = 0, 100
|
||||
timeout = timeout or 30000
|
||||
while not backend:isReady() do
|
||||
if waited >= timeout then return false end
|
||||
Citizen.Wait(step)
|
||||
waited = waited + step
|
||||
end
|
||||
return true
|
||||
end
|
||||
@@ -0,0 +1,746 @@
|
||||
// ---------------------------------------------------------------------------
|
||||
// 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();
|
||||
Reference in New Issue
Block a user