AzerothAI/scripts/ale-db-relay.mjs

import {execFileSync} from 'node:child_process';

const BASE=(process.env.AZEROTHAI_URL||'http://127.0.0.1:8787').replace(/\/+$/,'');
const TOKEN=process.env.ALE_BRIDGE_TOKEN||'';
const DB_PASSWORD=process.env.DOCKER_DB_ROOT_PASSWORD||'';
const DB_CONTAINER=process.env.AZEROTHAI_DB_CONTAINER||'ac-database';
const POLL_MS=Math.max(100,Number(process.env.AZEROTHAI_ALE_RELAY_POLL_MS||250));
const EVENT_BATCH=Math.max(1,Math.min(20,Number(process.env.AZEROTHAI_ALE_EVENT_BATCH||8)));
const EVENT_GAP_MS=Math.max(0,Number(process.env.AZEROTHAI_ALE_EVENT_GAP_MS||75));

if(!TOKEN)throw new Error('ALE_BRIDGE_TOKEN is required');
if(!DB_PASSWORD)throw new Error('DOCKER_DB_ROOT_PASSWORD is required');

let stopping=false;
let lastCleanup=0;
process.on('SIGINT',()=>{stopping=true;});
process.on('SIGTERM',()=>{stopping=true;});

function sleep(ms){return new Promise(r=>setTimeout(r,ms));}
function sqlText(value){
 const hex=Buffer.from(String(value??''),'utf8').toString('hex');
 return `CONVERT(UNHEX('${hex}') USING utf8mb4)`;
}
function mysql(sql){
 return execFileSync('docker',[
  'exec','-e',`MYSQL_PWD=${DB_PASSWORD}`,DB_CONTAINER,
  'mysql','-uroot','-N','-B','-e',sql
 ],{encoding:'utf8',stdio:['ignore','pipe','pipe']});
}

function ensureSchema(){
 mysql(`
CREATE TABLE IF NOT EXISTS acore_auth.azerothai_command_inbox (
 id VARCHAR(36) NOT NULL PRIMARY KEY,
 source MEDIUMTEXT NOT NULL,
 created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
 consumed_at TIMESTAMP(6) NULL,
 KEY idx_azerothai_command_inbox_pending (consumed_at,created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS acore_auth.azerothai_event_outbox (
 id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
 kind VARCHAR(100) NOT NULL,
 payload MEDIUMTEXT NOT NULL,
 created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
 delivered_at TIMESTAMP(6) NULL,
 KEY idx_azerothai_event_outbox_pending (delivered_at,id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS acore_auth.azerothai_result_outbox (
 seq BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
 command_id VARCHAR(36) NOT NULL,
 ok TINYINT(1) NOT NULL,
 result MEDIUMTEXT NOT NULL,
 created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
 delivered_at TIMESTAMP(6) NULL,
 UNIQUE KEY uq_azerothai_result_command (command_id),
 KEY idx_azerothai_result_outbox_pending (delivered_at,seq)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;`);
}

async function backend(path,init={}){
 const headers=new Headers(init.headers||{});
 headers.set('Authorization',`Bearer ${TOKEN}`);
 headers.set('Accept','application/json');
 const controller=new AbortController();
 const timer=setTimeout(()=>controller.abort(),5000);
 try{
  return await fetch(BASE+path,{...init,headers,signal:controller.signal});
 }finally{
  clearTimeout(timer);
 }
}

async function pullCommand(){
 const response=await backend('/api/bridge/ale/poll',{
  method:'POST',
  headers:{'Content-Type':'application/json'},
  body:'{}'
 });
 if(!response.ok)throw new Error(`poll HTTP ${response.status}: ${await response.text()}`);
 const id=response.headers.get('x-azerothai-command');
 const source=await response.text();
 if(!id)return false;
 mysql(`INSERT INTO acore_auth.azerothai_command_inbox(id,source)
 VALUES (${sqlText(id)},${sqlText(source)})
 ON DUPLICATE KEY UPDATE source=VALUES(source)`);
 return true;
}

function rows(sql){
 const out=mysql(sql).trim();
 if(!out)return [];
 return out.split('\n').filter(Boolean).map(line=>line.split('\t'));
}

async function deliverEvents(){
 // Reserve capacity for current snapshots so historical events cannot starve
 // live decisions. The remaining slots continue draining history in order.
 const fresh=rows(`SELECT id,HEX(kind),HEX(payload)
 FROM acore_auth.azerothai_event_outbox
 WHERE delivered_at IS NULL AND kind='player.snapshot'
 AND created_at >= NOW() - INTERVAL 90 SECOND
 ORDER BY id ASC LIMIT 64`);
 if(fresh.length){
  const payloads=fresh.map(([, ,hex])=>JSON.parse(Buffer.from(hex,'hex').toString('utf8')));
  const response=await backend('/api/bridge/ale/snapshots',{
   method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({payloads})
  });
  if(!response.ok)throw new Error(`snapshot batch HTTP ${response.status}: ${await response.text()}`);
  mysql(`UPDATE acore_auth.azerothai_event_outbox SET delivered_at=NOW(6) WHERE id IN (${fresh.map(row=>Number(row[0])).join(',')})`);
 }
 const batch=rows(`SELECT id,HEX(kind),HEX(payload)
 FROM acore_auth.azerothai_event_outbox
 WHERE delivered_at IS NULL
 ${fresh.length?`AND id NOT IN (${fresh.map(row=>Number(row[0])).join(',')})`:''}
 ORDER BY CASE WHEN kind IN ('player.permanent-death','bridge.ready') THEN 0 WHEN kind IN ('player.whisper','player.chat') AND created_at >= NOW() - INTERVAL 90 SECOND THEN 1 ELSE 2 END, id
 LIMIT ${EVENT_BATCH}`);
 for(const [id,kindHex,payloadHex] of batch){
  const kind=Buffer.from(kindHex||'','hex').toString('utf8');
  const payloadText=Buffer.from(payloadHex||'','hex').toString('utf8');
  let payload;
  try{payload=JSON.parse(payloadText);}catch{
   console.error(`Dropping malformed ALE event ${id}: invalid JSON payload`);
   mysql(`UPDATE acore_auth.azerothai_event_outbox SET delivered_at=NOW(6) WHERE id=${Number(id)}`);
   continue;
  }
  const response=await backend('/api/bridge/ale/event',{
   method:'POST',
   headers:{'Content-Type':'application/json'},
   body:JSON.stringify({kind,payload})
  });
  if(!response.ok)throw new Error(`event HTTP ${response.status}: ${await response.text()}`);
  mysql(`UPDATE acore_auth.azerothai_event_outbox SET delivered_at=NOW(6) WHERE id=${Number(id)}`);
  if(EVENT_GAP_MS)await sleep(EVENT_GAP_MS);
 }
 return fresh.length+batch.length;
}

async function deliverResults(){
 const batch=rows("SELECT seq,command_id,ok,HEX(result) FROM acore_auth.azerothai_result_outbox WHERE delivered_at IS NULL ORDER BY seq LIMIT 20");
 for(const [seq,commandId,ok,resultHex] of batch){
  const result=Buffer.from(resultHex||'','hex').toString('utf8').slice(0,16000);
  const response=await backend('/api/bridge/ale/result',{
   method:'POST',
   headers:{'Content-Type':'application/json'},
   body:JSON.stringify({id:commandId,ok:ok==='1',result})
  });
  if(!response.ok)throw new Error(`result HTTP ${response.status}: ${await response.text()}`);
  mysql(`UPDATE acore_auth.azerothai_result_outbox SET delivered_at=NOW(6) WHERE seq=${Number(seq)}`);
 }
 return batch.length;
}

function cleanup(){
 mysql(`
CREATE TABLE IF NOT EXISTS acore_auth.azerothai_snapshot_archive LIKE acore_auth.azerothai_event_outbox;
INSERT IGNORE INTO acore_auth.azerothai_snapshot_archive
 SELECT * FROM acore_auth.azerothai_event_outbox
 WHERE delivered_at IS NULL AND kind='player.snapshot'
 AND created_at < NOW() - INTERVAL 90 SECOND;
UPDATE acore_auth.azerothai_event_outbox e
 JOIN acore_auth.azerothai_snapshot_archive a ON a.id=e.id
 SET e.delivered_at=NOW(6) WHERE e.delivered_at IS NULL;
UPDATE acore_auth.azerothai_event_outbox
 SET delivered_at=NOW(6)
 WHERE delivered_at IS NULL
   AND kind IN ('player.combat.enter','player.combat.leave')
   AND created_at < NOW() - INTERVAL 2 MINUTE;
DELETE FROM acore_auth.azerothai_command_inbox WHERE consumed_at IS NOT NULL AND consumed_at < NOW() - INTERVAL 1 DAY;
DELETE FROM acore_auth.azerothai_event_outbox WHERE delivered_at IS NOT NULL AND delivered_at < NOW() - INTERVAL 1 DAY;
DELETE FROM acore_auth.azerothai_result_outbox WHERE delivered_at IS NOT NULL AND delivered_at < NOW() - INTERVAL 1 DAY;`);
}

ensureSchema();
console.log(`AzerothAI ALE DB relay started: ${BASE}, poll ${POLL_MS}ms`);

while(!stopping){
 try{
  if(Date.now()-lastCleanup>60_000){cleanup();lastCleanup=Date.now();}
  await deliverEvents();
  await deliverResults();
  await pullCommand();
  await sleep(POLL_MS);
 }catch(error){
  console.error(new Date().toISOString(),error instanceof Error?error.message:String(error));
  await sleep(1000);
 }
}

console.log('AzerothAI ALE DB relay stopped.');