Compare commits

..

7 Commits

83 changed files with 2997 additions and 1384 deletions
+1
View File
@@ -0,0 +1 @@
@p42:registry=https://gitea.internike.com/api/packages/P42/npm/
-70
View File
@@ -1,70 +0,0 @@
export const construct = (redisCnx) => {
const tickMs = redisCnx.gpsSrv?.getGpsSettings().collisionTickMs ?? 100
// Interval always runs; tickArena no-ops until LIVE (see gpsServer.tickArena)
setInterval(() => {
redisCnx.gpsSrv?.tickArena()
}, tickMs)
}
export const methods = {
handleLifecycleEvent(msg) {
const srv = this.gpsSrv
if(!srv) return
if(msg.eventType === 'onYourMarks') {
srv.onYourMarks(msg.payload ?? {}).catch(err => {
console.error(`[${this.redisId}] onYourMarks failed:`, err)
srv.publishReadyToStart({ success: false, err: err.message ?? 'onYourMarks failed' })
})
return
}
if(msg.eventType === 'bigBang') {
srv.onBigBang(msg.payload ?? {})
return
}
},
handleAgentEvent(msg) {
const agentId = msg.sender
if(!agentId || typeof(agentId) !== 'string') {
console.warn(`[${this.redisId}] Agent event without sender`)
return
}
if(msg.eventType === 'change') {
const newVector = msg.payload?.newVector
if(!newVector || typeof(newVector.x) !== 'number' || typeof(newVector.y) !== 'number' || typeof(newVector.z) !== 'number') {
console.warn(`[${this.redisId}] Invalid newVector from ${agentId}`)
return
}
const newPosition = msg.payload?.newPosition ?? null
this.gpsSrv.onVectorChange(agentId, newVector, newPosition)
return
}
if(msg.eventType === 'remove') {
this.gpsSrv.onAgentRemove(agentId)
return
}
},
dispatchArenaMessage(msg, chan) {
const gps = this.config.gps
if(!gps || !this.gpsSrv) return(false)
if(this.matchesChan(chan, gps.lifecycle?.arenaChannel ?? 'arena:lifecycle')) {
this.handleLifecycleEvent(msg)
return(true)
}
if(this.matchesChan(chan, gps.agentVectorChangeChannel)) {
this.handleAgentEvent(msg)
return(true)
}
return(false)
},
}
-5
View File
@@ -1,5 +0,0 @@
export function dispatchMessage(redisCnx, msg, chan) {
if(!redisCnx.config.gps || typeof(redisCnx.dispatchArenaMessage) !== 'function') return
redisCnx.dispatchArenaMessage(msg, chan)
}
-11
View File
@@ -1,11 +0,0 @@
import { methods as arenaMethods, construct as arenaConstruct } from './arenaHandlers.js'
import { dispatchMessage } from './dispatch.js'
export const afterLoginMethods = [
arenaConstruct,
]
export const meshActions = {
...arenaMethods,
}
export { dispatchMessage }
-28
View File
@@ -1,28 +0,0 @@
export function dispatchMessage(redisCnx, msg, chan) {
const gps = redisCnx.config.gps
if(!gps?.gpsActionsChannel) return
const actionsChan = redisCnx.fullChan(gps.gpsActionsChannel)
if(chan != actionsChan) return
const action = msg.action
if(!action || typeof(action) !== 'string') {
console.warn(`[${redisCnx.redisId}] Ignoring message without action on ${chan}`)
return
}
const handler = redisCnx['action_'+action]
if(typeof(handler) != 'function') {
if(redisCnx.debug) console.warn(`[${redisCnx.redisId}] Unknown action ${action} on ${chan}`)
return
}
const payload = ('payload' in msg) ? msg.payload : null
const reqid = ('reqid' in msg) ? msg.reqid.substr(0, 50) : null
const sender = msg.sender || null
const roles = Array.isArray(msg.roles) ? msg.roles : ['*']
if(redisCnx.debug) console.log(`[${redisCnx.redisId}] Dispatching action ${action} from ${sender}`)
handler.call(redisCnx, action, payload, reqid, sender, roles)
}
-12
View File
@@ -1,12 +0,0 @@
import { methods as utilities, construct as utilitiesConstruct } from './utilities.js'
import { methods as positions } from './positions.js'
import { dispatchMessage } from './dispatch.js'
export const afterLoginMethods = [
utilitiesConstruct,
]
export const meshActions = {
...utilities,
...positions,
}
export { dispatchMessage }
-152
View File
@@ -1,152 +0,0 @@
import { publishActionReply, parseSimTime } from '../../actionsHelper.js'
export const methods = {
/* Event-Rx:
{
"action": "GETAGENTPOSITION",
"reqid": "6az5e4r6a",
"payload": {
"agentId": "agent42",
"t": 12.5
}
}
Event-Tx:
{
"action": "GETAGENTPOSITION",
"success": true,
"reqid": "6az5e4r6a",
"payload": {
"agent": {
"id": "agent42",
"position": { "x": 1, "y": 2, "z": 3 },
"vector": { "x": 0, "y": 0, "z": 0 },
"since": 0,
"generation": 2,
"t": 12.5
}
}
}
*/
async action_GETAGENTPOSITION(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.gps.gpsActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
if(!this.gpsSrv.isLive()) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Simulation not live',
} })
return
}
const agentId = payload?.agentId
if(!agentId || typeof(agentId) !== 'string') {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Missing or invalid agentId',
} })
return
}
const at = parseSimTime(payload, () => this.gpsSrv.now())
if(at === null) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Invalid simulation time',
} })
return
}
const agent = this.gpsSrv.getAgentPosition(agentId, at)
if(!agent) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: `Unknown agent: ${agentId}`,
} })
return
}
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: { agent },
} })
},
/* Event-Rx:
{
"action": "GETAGENTSINPRISM",
"reqid": "6az5e4r6a",
"payload": {
"prism": {
"xMin": -10, "xMax": 10,
"yMin": -10, "yMax": 10,
"zMin": 0, "zMax": 5
},
"t": 0
}
}
*/
async action_GETAGENTSINPRISM(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.gps.gpsActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
if(!this.gpsSrv.isLive()) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Simulation not live',
} })
return
}
const prism = payload?.prism
if(!this.gpsSrv.isValidPrism(prism)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Missing or invalid prism bounds',
} })
return
}
const at = parseSimTime(payload, () => this.gpsSrv.now())
if(at === null) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Invalid simulation time',
} })
return
}
const agents = this.gpsSrv.getAgentsInPrism(prism, at)
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: {
agents,
t: at,
},
} })
},
}
-109
View File
@@ -1,109 +0,0 @@
import { publishActionReply } from '../../actionsHelper.js'
export const construct = (redisCnx) => {
// console.log('Hello after login from utilities...')
// redisCnx.v42=0
// setInterval(redisCnx.move4243.bind(redisCnx), 200)
}
export const methods = {
/* Event-Rx:
{
"action": "TIME"
"reqid": "6az5e4r6a"
}
Event-Tx:
{
"action": "TIME",
"success": true,
"payload" : {
gpsTime: "2022-09-01T14:42:22.603Z",
redisTime: "2022-09-01T14:42:22.603Z"
},
"reqid": "6az5e4r6a"
}
*/
async action_TIME(action, payload, reqid, sender, roles){
publishActionReply(this, {
action,
reqid,
sender,
replyChannel: this.config.gps.gpsActionsReply,
reply: {
success: true,
payload: {
gpsTime: new Date().toISOString(),
redisTime: await this.redisClient.time(),
},
},
})
},
/* Event-Rx:
{
"action": "RELOADCONFIG"
"reqid": "6az5e4r6a"
}
Event-Tx:
{
"action": "RELOADCONFIG",
"success": true,
"reqid": "6az5e4r6a"
}
*/
async action_RELOADCONFIG(action, payload, reqid, sender, roles){
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.gps.gpsActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
this.reloadAccessRights()
publishActionReply(this, { ...replyOpts, reply: {
success: true,
} })
},
/* Event-Rx:
{
"action": "GETCONFIG"
"reqid": "6az5e4r6a"
}
Event-Tx:
{
"action": "GETCONFIG",
"success": true,
"reqid": "6az5e4r6a",
payload: { ...the access rights, and roles... }
}
*/
async action_GETCONFIG(action, payload, reqid, sender, roles){
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.gps.gpsActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: this.getAccessRights(),
} })
},
}
-22
View File
@@ -1,22 +0,0 @@
export function publishActionReply(redisCnx, options) {
const {
action,
reqid,
sender,
reply,
replyChannel,
senderId = 'gps',
} = options
reply.action = action
reply.sender = senderId
if(reqid) reply.reqid = reqid
const chan = replyChannel.replace(/\[UID\]/g, sender)
redisCnx.redisPublish(chan, reply)
}
export function parseSimTime(payload, fallbackFn) {
if(payload?.t != null && typeof(payload.t) === 'number' && !Number.isNaN(payload.t)) return(payload.t)
if(payload?.at != null && typeof(payload.at) === 'number' && !Number.isNaN(payload.at)) return(payload.at)
return(fallbackFn())
}
+11 -11
View File
@@ -1,23 +1,23 @@
export class AgentStore {
constructor(systemCnx, storage, debug = false) {
constructor(systemCnx, gpsStorage, debug = false) {
this.cnx = systemCnx
this.storage = storage
this.gpsStorage = gpsStorage
this.debug = debug
}
agentHashKey(agentId) {
return(this.storage.agentHashKey.replace(/\[UID\]/g, agentId))
return(this.gpsStorage.agentHashKey.replace(/\[UID\]/g, agentId))
}
async clearAll() {
try {
const ids = await this.cnx.redisSmembers(this.storage.agentsIndexKey)
const ids = await this.cnx.redisSmembers(this.gpsStorage.agentsIndexKey)
for(const id of ids) {
await this.cnx.redisDel(this.agentHashKey(id))
}
await this.cnx.redisDel(this.storage.agentsIndexKey)
await this.cnx.redisDel(this.gpsStorage.agentsIndexKey)
if(this.debug) console.log(`[GPS] Cleared system agent store (${ids.length} agent(s))`)
} catch(err) {
console.error('[GPS] Failed to clear system agent store:', err)
@@ -37,11 +37,11 @@ export class AgentStore {
simulationId,
}
await this.cnx.redisHset(this.agentHashKey(agent.id), 'segment', record)
await this.cnx.redisSadd(this.storage.agentsIndexKey, agent.id)
await this.cnx.redisSadd(this.gpsStorage.agentsIndexKey, agent.id)
await this.cnx.redisXadd(
this.storage.positionsStream,
this.gpsStorage.positionsStream,
record,
this.storage.streamMaxLen ?? ''
this.gpsStorage.streamMaxLen ?? ''
)
if(this.debug) console.log(`[GPS] Exported segment ${agent.id} (${eventType})`)
} catch(err) {
@@ -57,11 +57,11 @@ export class AgentStore {
t: simT,
}
await this.cnx.redisDel(this.agentHashKey(agentId))
await this.cnx.redisSrem(this.storage.agentsIndexKey, agentId)
await this.cnx.redisSrem(this.gpsStorage.agentsIndexKey, agentId)
await this.cnx.redisXadd(
this.storage.positionsStream,
this.gpsStorage.positionsStream,
record,
this.storage.streamMaxLen ?? ''
this.gpsStorage.streamMaxLen ?? ''
)
if(this.debug) console.log(`[GPS] Exported remove ${agentId}`)
} catch(err) {
+4 -4
View File
@@ -1,9 +1,9 @@
export class ArenaAgentLoader {
constructor(arenaCnx, storage, debug = false) {
constructor(arenaCnx, arenaStorage, debug = false) {
this.cnx = arenaCnx
this.storage = storage
this.arenaStorage = arenaStorage
this.debug = debug
}
@@ -45,11 +45,11 @@ export class ArenaAgentLoader {
async #listAgentIds(expectedIds = null) {
if(Array.isArray(expectedIds) && expectedIds.length) return([...expectedIds])
return(await this.cnx.redisSmembers(this.storage.agentsIndexKey))
return(await this.cnx.redisSmembers(this.arenaStorage.agentsIndexKey))
}
async #loadAgentFromHash(agentId) {
const key = this.storage.agentHashKey.replace(/\[UID\]/g, agentId)
const key = this.arenaStorage.agentHashKey.replace(/\[UID\]/g, agentId)
const positionRaw = await this.cnx.redisHget(key, 'position')
const vectorRaw = await this.cnx.redisHget(key, 'vector')
let position = this.#parseHashField(positionRaw)
+81 -47
View File
@@ -8,16 +8,17 @@ import {
positionAt,
needsPrismRefresh,
advanceAgentSegment,
} from './actions/arena/worldline.js'
} from './handlers/arena/worldline.js'
export class gpsServer {
constructor(configHelper, allRediscnx, debug) {
this.configHelper = configHelper
this.gpsConfig = configHelper.config
this.rootConfig = configHelper.config
this.gpsConfig = configHelper.config.gps ?? {}
this.allRediscnx = allRediscnx
this.debug = debug
this.accessRights = new AccesRights(this.gpsConfig, debug)
this.accessRights = new AccesRights(this.rootConfig, debug)
this.agents = new Map()
this.registry = new CollisionRegistry()
this.arenaCnx = null
@@ -28,51 +29,38 @@ export class gpsServer {
this.state = SimState.IDLE
this.simulationId = null
this.bigBangEpoch = null
this.resumeEpoch = null
this.pausedAt = null
this.ignoredChangeCount = 0
}
getGpsSettings() {
const gps = this.gpsConfig.gps ?? {}
return({
nearMissDistance: gps.nearMissDistance ?? 1,
prismTimeHeight: gps.prismTimeHeight ?? 60,
collisionTickMs: gps.collisionTickMs ?? 100,
prismRefreshLeadSeconds: gps.prismRefreshLeadSeconds ?? 1,
})
}
getLifecycleSettings() {
const gps = this.gpsConfig.gps ?? {}
return({
arenaChannel: gps.lifecycle?.arenaChannel ?? 'arena:lifecycle',
godsReadyChannel: gps.lifecycle?.godsReadyChannel ?? 'arena:gods:ready',
senderId: gps.senderId ?? 'gps',
})
}
getArenaStorageSettings() {
const gps = this.gpsConfig.gps ?? {}
return({
agentHashKey: gps.arenaStorage?.agentHashKey ?? 'arena:agents:[UID]',
agentsIndexKey: gps.arenaStorage?.agentsIndexKey ?? 'arena:agents',
})
}
isLive() {
return(this.state === SimState.LIVE)
}
isPaused() {
return(this.state === SimState.PAUSED)
}
canQuerySim() {
return(this.isLive() || this.isPaused())
}
isPrepare() {
return(this.state === SimState.PREPARE)
}
simNow() {
if(this.bigBangEpoch === null) return(null)
if(this.resumeEpoch !== null) {
return(this.pausedAt + (performance.now() - this.resumeEpoch) / 1000)
}
return((performance.now() - this.bigBangEpoch) / 1000)
}
now() {
if(this.isLive()) return(this.simNow())
if(this.isPaused()) return(this.pausedAt)
if(this.isPrepare()) return(0)
return(null)
}
@@ -89,9 +77,9 @@ export class gpsServer {
}
initAgentStore() {
const storage = this.gpsConfig.gps?.storage
if(storage && this.systemCnx) {
this.agentStore = new AgentStore(this.systemCnx, storage, this.debug)
const gpsStorage = this.gpsConfig.GPSstorage
if(gpsStorage && this.systemCnx) {
this.agentStore = new AgentStore(this.systemCnx, gpsStorage, this.debug)
}
}
@@ -100,7 +88,10 @@ export class gpsServer {
this.arenaCnxs.push(cnx)
if(!this.arenaCnx || cnx.redisConfig.role === 'primary') {
this.arenaCnx = cnx
this.arenaLoader = new ArenaAgentLoader(cnx, this.getArenaStorageSettings(), this.debug)
const arenaStorage = this.gpsConfig.arenaStorage
if(arenaStorage) {
this.arenaLoader = new ArenaAgentLoader(cnx, arenaStorage, this.debug)
}
}
}
@@ -124,7 +115,7 @@ export class gpsServer {
}
getAgentPosition(agentId, at = null) {
if(!this.isLive()) return(null)
if(!this.canQuerySim()) return(null)
const agent = this.agents.get(agentId)
if(!agent) return(null)
const t = at ?? this.now()
@@ -154,7 +145,7 @@ export class gpsServer {
}
getAgentsInPrism(prism, at = null) {
if(!this.isLive()) return([])
if(!this.canQuerySim()) return([])
const t = at ?? this.now()
if(t === null) return([])
const agents = []
@@ -172,13 +163,15 @@ export class gpsServer {
this.registry = new CollisionRegistry()
this.simulationId = null
this.bigBangEpoch = null
this.resumeEpoch = null
this.pausedAt = null
this.ignoredChangeCount = 0
this.state = SimState.IDLE
await this.agentStore?.clearAll()
}
runInitialPairScan() {
const { nearMissDistance, prismTimeHeight } = this.getGpsSettings()
const { nearMissDistance, prismTimeHeight } = this.gpsConfig
const ids = [...this.agents.keys()]
for(let i = 0; i < ids.length; i++) {
for(let j = i + 1; j < ids.length; j++) {
@@ -197,10 +190,9 @@ export class gpsServer {
async publishReadyToStart(result) {
if(!this.arenaCnx) return
const { godsReadyChannel, senderId } = this.getLifecycleSettings()
await this.arenaCnx.redisPublish(godsReadyChannel, {
await this.arenaCnx.redisPublish(this.gpsConfig.lifecycle.godsReadyChannel, {
eventType: 'readyToStart',
sender: senderId,
sender: this.gpsConfig.senderId,
payload: {
success: result.success,
simulationId: this.simulationId,
@@ -269,6 +261,8 @@ export class gpsServer {
}
this.bigBangEpoch = performance.now()
this.resumeEpoch = null
this.pausedAt = null
this.state = SimState.LIVE
for(const agent of this.agents.values()) {
@@ -278,6 +272,46 @@ export class gpsServer {
if(this.debug) console.log(`[GPS] LIVE: bigBangEpoch set, simulationId=${this.simulationId}`)
}
onSimulationPaused(payload = {}) {
if(this.state === SimState.IDLE || this.state === SimState.PAUSED) return
if(payload.simulationId && this.simulationId && payload.simulationId !== this.simulationId) {
console.error('[GPS] simulationPaused rejected: simulationId mismatch')
return
}
const t = (typeof(payload.t) === 'number' && !Number.isNaN(payload.t))
? payload.t
: this.now()
this.pausedAt = t
this.state = SimState.PAUSED
if(this.debug) console.log(`[GPS] PAUSED at t=${t}, simulationId=${this.simulationId}`)
}
async onSimulationStopped(payload = {}) {
if(this.state === SimState.IDLE) return
if(payload.simulationId && this.simulationId && payload.simulationId !== this.simulationId) {
console.error('[GPS] simulationStopped rejected: simulationId mismatch')
return
}
if(this.debug) console.log(`[GPS] STOPPED (reset), simulationId=${this.simulationId}`)
await this.resetSimulation()
}
onSimulationResumed(payload = {}) {
if(this.state !== SimState.PAUSED) {
console.error(`[GPS] simulationResumed rejected: expected PAUSED, got ${this.state}`)
return
}
if(payload.simulationId && this.simulationId && payload.simulationId !== this.simulationId) {
console.error('[GPS] simulationResumed rejected: simulationId mismatch')
return
}
this.resumeEpoch = performance.now()
this.state = SimState.LIVE
if(this.debug) console.log(`[GPS] RESUMED at t=${this.pausedAt}, simulationId=${this.simulationId}`)
}
upsertAgent(agentId, newVector, newPosition = null) {
const now = this.now()
let agent = this.agents.get(agentId)
@@ -312,7 +346,7 @@ export class gpsServer {
const agent = this.agents.get(agentId)
if(!agent) return(false)
const { prismTimeHeight, prismRefreshLeadSeconds } = this.getGpsSettings()
const { prismTimeHeight, prismRefreshLeadSeconds } = this.gpsConfig
const now = this.now()
if(!needsPrismRefresh(agent, now, prismTimeHeight, prismRefreshLeadSeconds)) return(false)
@@ -332,7 +366,7 @@ export class gpsServer {
const changed = this.agents.get(changedAgentId)
if(!changed) return([])
const { nearMissDistance, prismTimeHeight } = this.getGpsSettings()
const { nearMissDistance, prismTimeHeight } = this.gpsConfig
const now = this.now()
const hits = []
for(const [otherId, other] of this.agents) {
@@ -401,14 +435,14 @@ export class gpsServer {
async publishProximityBatch(targetAgentId, pairs) {
if(!this.arenaCnx || !pairs.length) return
const chan = this.arenaCnx.config.gps.collisionsChannel.replace(/\[UID\]/g, targetAgentId)
const chan = this.gpsConfig.collisionsChannel.replace(/\[UID\]/g, targetAgentId)
await this.arenaCnx.redisPublish(chan, {
eventType: 'proximity',
payload: {
pairs,
simulationId: this.simulationId,
},
sender: this.getLifecycleSettings().senderId,
sender: this.gpsConfig.senderId,
})
}
@@ -433,12 +467,12 @@ export class gpsServer {
async reloadAccessRights() {
await this.configHelper.refreshAccessRights()
this.gpsConfig.accessRights = this.configHelper.config.accessRights
this.accessRights.refreshAccessRights(this.gpsConfig)
this.rootConfig.accessRights = this.configHelper.config.accessRights
this.accessRights.refreshAccessRights(this.rootConfig)
}
getAccessRights() {
return(this.gpsConfig.accessRights)
return(this.rootConfig.accessRights)
}
}
+27
View File
@@ -0,0 +1,27 @@
export const eventHandlers = {
'arena:agents:*': {
change(msg, chan, sender, cnxId) {
const agentId = msg.sender
if(!agentId || typeof(agentId) !== 'string') {
console.warn(`[${this.redisId}] Agent event without sender`)
return
}
const newVector = msg.payload?.newVector
if(!newVector || typeof(newVector.x) !== 'number' || typeof(newVector.y) !== 'number' || typeof(newVector.z) !== 'number') {
console.warn(`[${this.redisId}] Invalid newVector from ${agentId}`)
return
}
const newPosition = msg.payload?.newPosition ?? null
this.gpsSrv?.onVectorChange(agentId, newVector, newPosition)
},
remove(msg, chan, sender, cnxId) {
const agentId = msg.sender
if(!agentId || typeof(agentId) !== 'string') {
console.warn(`[${this.redisId}] Agent event without sender`)
return
}
this.gpsSrv?.onAgentRemove(agentId)
},
},
}
+18
View File
@@ -0,0 +1,18 @@
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
import * as lifecycle from './lifecycle.js'
import * as agentMotion from './agentMotion.js'
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([lifecycle, agentMotion])
export { actionHandlers, afterLogin }
export const dispatchMessage = createDispatchMessage({
eventHandlers,
actionRules(redisCnx) {
const gps = redisCnx.config.gps ?? {}
const arenaChannel = gps.arenaActionsChannel
return({
channels: arenaChannel ? [arenaChannel] : [],
})
},
})
+36
View File
@@ -0,0 +1,36 @@
export function construct(redisCnx) {
const tickMs = redisCnx.gpsSrv?.gpsConfig.collisionTickMs ?? 100
setInterval(() => {
redisCnx.gpsSrv?.tickArena()
}, tickMs)
}
export const eventHandlers = {
'arena:lifecycle': {
onYourMarks(msg, chan, sender, cnxId) {
const srv = this.gpsSrv
if(!srv) return
srv.onYourMarks(msg.payload ?? {}).catch(err => {
console.error(`[${this.redisId}] onYourMarks failed:`, err)
srv.publishReadyToStart({ success: false, err: err.message ?? 'onYourMarks failed' })
})
},
bigBang(msg, chan, sender, cnxId) {
this.gpsSrv?.onBigBang(msg.payload ?? {})
},
simulationPaused(msg, chan, sender, cnxId) {
this.gpsSrv?.onSimulationPaused(msg.payload ?? {})
},
simulationResumed(msg, chan, sender, cnxId) {
this.gpsSrv?.onSimulationResumed(msg.payload ?? {})
},
simulationStopped(msg, chan, sender, cnxId) {
const srv = this.gpsSrv
if(!srv) return
srv.onSimulationStopped(msg.payload ?? {}).catch(err => {
console.error(`[${this.redisId}] simulationStopped failed:`, err)
})
},
},
}
@@ -19,7 +19,7 @@ export function positionAt(agent, t) {
})
}
function distanceBetween(agentA, agentB, t) {
export function distanceBetween(agentA, agentB, t) {
const a = positionAt(agentA, t)
const b = positionAt(agentB, t)
const dx = a.x - b.x
+16
View File
@@ -0,0 +1,16 @@
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
import * as utilities from './utilities.js'
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([utilities])
export { actionHandlers, afterLogin }
export const dispatchMessage = createDispatchMessage({
eventHandlers,
actionRules(redisCnx) {
const gps = redisCnx.config.gps ?? {}
return({
channels: [gps.ActionsChannel].filter(Boolean),
})
},
})
+35
View File
@@ -0,0 +1,35 @@
import { replyToAction } from '../../../bus/publishActionReply.js'
export const actions = {
async action_TIME(action, payload, reqid, sender, cnxId, roles) {
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: {
gpsTime: new Date().toISOString(),
redisTime: await this.redisClient.time(),
},
})
},
async action_RELOADCONFIG(action, payload, reqid, sender, cnxId, roles) {
this.reloadAccessRights()
replyToAction(this, { action, reqid, sender, cnxId, success: true })
},
async action_GETCONFIG(action, payload, reqid, sender, cnxId, roles) {
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: this.getAccessRights(),
})
},
}
+6 -2
View File
@@ -3,10 +3,11 @@ import yargs from 'yargs/yargs'
import { hideBin } from 'yargs/helpers'
import 'node:process'
import {RedisConnexion} from '../redisConnexion.js'
import { busReplyRoute } from '../bus/publishActionReply.js'
import {configHelper} from '../configHelper.js'
import {gpsServer} from './gpsServer.js'
import * as systemMesh from './actions/system/index.js'
import * as arenaMesh from './actions/arena/index.js'
import * as systemMesh from './handlers/system/index.js'
import * as arenaMesh from './handlers/arena/index.js'
const meshModules = {
system: systemMesh,
@@ -53,6 +54,7 @@ let cfgh = new configHelper({
function meshRedisConns(mesh, meshName, debug, rootConfig) {
const { redis, ...meshConfig } = mesh
const busRoute = busReplyRoute(rootConfig.gps, meshName)
return redis.map(cfg =>
new RedisConnexion({
debug,
@@ -60,6 +62,8 @@ function meshRedisConns(mesh, meshName, debug, rootConfig) {
redisId: cfg.redisId,
meshName,
meshModule: meshModules[meshName],
senderId: busRoute?.senderId,
actionsReply: busRoute?.actionsReply,
})
)
}
+1
View File
@@ -2,6 +2,7 @@ export const SimState = {
IDLE: 'idle',
PREPARE: 'prepare',
LIVE: 'live',
PAUSED: 'paused',
}
export function validateAgentSets(expected, found) {
+15 -3
View File
@@ -1,16 +1,28 @@
#!/bin/sh
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
# shellcheck source=../lib/resolveConfigPath.sh
. "$SCRIPT_DIR/../lib/resolveConfigPath.sh"
CONFIG="$(resolveConfigPath "$SCRIPT_DIR" "${1:-${CONFIG:-../config.json}}")"
set -a
. /etc/p42/secrets.env
set +a
daemon=p42Gps
logfile=gps.log
if [ ! -f "$CONFIG" ]; then
echo "Config file not found: $CONFIG" >&2
exit 1
fi
pid=$(pgrep -f "$daemon")
if [ -z "$pid" ]
then
node "${daemon}.js" --debug > "$logfile" 2>&1 &
node "${daemon}.js" --config "$CONFIG" --debug > "$logfile" 2>&1 &
pid=$!
sleep 1
@@ -18,16 +30,16 @@ then
if kill -0 "$pid" 2>/dev/null
then
echo ""
echo "$daemon is now running with PID=$pid"
echo "$daemon is now running with PID=$pid (config=$CONFIG)"
echo ""
else
echo ""
echo "Failed to start $daemon. Check gps.log"
echo ""
exit 1
fi
else
echo ""
echo "$daemon is already running with PID=$pid"
echo ""
fi
@@ -1,22 +1,23 @@
const RESERVED_HASH_FIELDS = new Set(['position', 'vector', 'speed', 'segment'])
export class ArenaGroom {
constructor(arenaCnx, storage, debug = false) {
constructor(arenaCnx, arenaStorage, debug = false) {
this.cnx = arenaCnx
this.storage = storage
this.arenaStorage = arenaStorage
this.debug = debug
}
agentHashKey(agentId) {
return(this.storage.agentHashKey.replace(/\[UID\]/g, agentId))
return(this.arenaStorage.agentHashKey.replace(/\[UID\]/g, agentId))
}
async clearArena() {
const ids = await this.cnx.redisSmembers(this.storage.agentsIndexKey)
const ids = await this.cnx.redisSmembers(this.arenaStorage.agentsIndexKey)
for(const id of ids) {
await this.cnx.redisDel(this.agentHashKey(id))
}
await this.cnx.redisDel(this.storage.agentsIndexKey)
await this.cnx.redisDel(this.arenaStorage.agentsIndexKey)
if(this.debug) console.log(`[Maestro] Cleared arena store (${ids.length} agent(s))`)
}
@@ -25,7 +26,11 @@ export class ArenaGroom {
const key = this.agentHashKey(agent.id)
await this.cnx.redisHset(key, 'position', agent.position)
await this.cnx.redisHset(key, 'vector', agent.vector)
await this.cnx.redisSadd(this.storage.agentsIndexKey, agent.id)
for(const [field, value] of Object.entries(agent.store ?? {})) {
if(RESERVED_HASH_FIELDS.has(field)) continue
await this.cnx.redisHset(key, field, value)
}
await this.cnx.redisSadd(this.arenaStorage.agentsIndexKey, agent.id)
}
if(this.debug) console.log(`[Maestro] Groomed ${agents.length} agent(s) into arena store`)
}
+17
View File
@@ -0,0 +1,17 @@
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
import * as prepare from './prepare.js'
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([prepare])
export { actionHandlers, afterLogin }
export const dispatchMessage = createDispatchMessage({
eventHandlers,
actionRules(redisCnx) {
const maestro = redisCnx.config.maestro ?? {}
const arenaChannel = maestro.arenaActionsChannel
return({
channels: arenaChannel ? [arenaChannel] : [],
})
},
})
+9
View File
@@ -0,0 +1,9 @@
export const eventHandlers = {
'arena:gods:ready': {
readyToStart(msg, chan, sender, cnxId) {
if(!this.maestroSrv) return
this.maestroSrv.handlePrepareAck(msg, chan)
},
},
}
+17
View File
@@ -0,0 +1,17 @@
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
import * as simulation from './simulation.js'
import * as utilities from './utilities.js'
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([simulation, utilities])
export { actionHandlers, afterLogin }
export const dispatchMessage = createDispatchMessage({
eventHandlers,
actionRules(redisCnx) {
const maestro = redisCnx.config.maestro ?? {}
return({
channels: [maestro.ActionsChannel].filter(Boolean),
})
},
})
+119
View File
@@ -0,0 +1,119 @@
import { replyToAction } from '../../../bus/publishActionReply.js'
import { isValidUuid } from '../../simRepository.js'
export const actions = {
async action_STARTSIMULATION(action, payload, reqid, sender, cnxId, roles) {
if(!isValidUuid(sender)) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid sender (user UUID)' })
return
}
if(!payload?.simulationUuid) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing simulationUuid' })
return
}
const result = await this.maestroSrv.startSimulation(sender, payload)
if(!result.ok) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: result.err })
return
}
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: {
simulationId: result.simulationId,
keyframeId: result.keyframeId,
infraId: result.infraId,
agentIds: result.agentIds,
resumed: result.resumed ?? false,
state: result.state ?? null,
},
})
},
async action_PAUSESIMULATION(action, payload, reqid, sender, cnxId, roles) {
if(!isValidUuid(sender)) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid sender (user UUID)' })
return
}
if(!payload?.simulationUuid) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing simulationUuid' })
return
}
const result = await this.maestroSrv.pauseSimulation(sender, payload)
if(!result.ok) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: result.err })
return
}
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: {
simulationId: result.simulationId,
t: result.t,
},
})
},
async action_STOPSIMULATION(action, payload, reqid, sender, cnxId, roles) {
if(!isValidUuid(sender)) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid sender (user UUID)' })
return
}
if(!payload?.simulationUuid) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing simulationUuid' })
return
}
const result = await this.maestroSrv.stopSimulation(sender, payload)
if(!result.ok) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: result.err })
return
}
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: { simulationId: result.simulationId },
})
},
async action_GETSIMULATIONSSTATUS(action, payload, reqid, sender, cnxId, roles) {
if(!isValidUuid(sender)) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid sender (user UUID)' })
return
}
const result = await this.maestroSrv.getSimulationsStatus(sender)
if(!result.ok) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: result.err })
return
}
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: result.simulations,
})
},
}
+21
View File
@@ -0,0 +1,21 @@
import { replyToAction } from '../../../bus/publishActionReply.js'
export const actions = {
async action_RELOADCONFIG(action, payload, reqid, sender, cnxId, roles) {
this.reloadAccessRights()
replyToAction(this, { action, reqid, sender, cnxId, success: true })
},
async action_GETCONFIG(action, payload, reqid, sender, cnxId, roles) {
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: this.getAccessRights(),
})
},
}
+401
View File
@@ -0,0 +1,401 @@
import { AccesRights } from '../accesRights.js'
import { MySQLClient } from '@p42/p42modules'
import { SimRepository } from './simRepository.js'
import { ArenaGroom } from './arenaGroom.js'
import { MaestroState } from './orchestrationState.js'
import { PrepareQuorum } from './prepareQuorum.js'
import { buildPrepareQuorum } from './primordialDaemons.js'
export class maestroServer {
constructor(configHelper, allRediscnx, debug) {
this.configHelper = configHelper
this.rootConfig = configHelper.config
this.maestroConfig = configHelper.config.maestro ?? {}
this.gpsConfig = configHelper.config.gps ?? {}
this.allRediscnx = allRediscnx
this.debug = debug
this.accessRights = new AccesRights(this.rootConfig, debug)
this.arenaCnx = null
this.arenaCnxs = []
this.systemCnx = null
this.db = null
this.simRepo = null
this.arenaGroom = null
this.prepareQuorum = null
this.orchestrationState = MaestroState.IDLE
this.simulationId = null
this.agentIds = []
this.keyframeId = null
this.infraId = null
this.bigBangEpoch = null
this.resumeEpoch = null
this.pausedAt = null
}
isIdle() {
return(this.orchestrationState === MaestroState.IDLE)
}
isLive() {
return(this.orchestrationState === MaestroState.LIVE)
}
isPaused() {
return(this.orchestrationState === MaestroState.PAUSED)
}
simNow() {
if(this.bigBangEpoch === null) return(null)
if(this.resumeEpoch !== null) {
return(this.pausedAt + (performance.now() - this.resumeEpoch) / 1000)
}
return((performance.now() - this.bigBangEpoch) / 1000)
}
async init() {
const mysqlCfg = this.rootConfig.mysql
if(!mysqlCfg) {
console.error('[Maestro] Missing mysql config')
return(false)
}
this.db = await MySQLClient.createPool(mysqlCfg)
this.simRepo = new SimRepository(this.db, MySQLClient.resolveDatabases(mysqlCfg), this.debug)
if(this.debug) console.log('[Maestro] MySQL pool ready')
return(true)
}
refreshArenaGroom() {
const arenaStorage = this.gpsConfig.arenaStorage
if(this.arenaCnx && arenaStorage) {
this.arenaGroom = new ArenaGroom(
this.arenaCnx,
arenaStorage,
this.debug
)
}
}
refreshPrepareQuorum() {
if(!this.arenaCnx) return
this.prepareQuorum = new PrepareQuorum({
ackChannel: this.maestroConfig.lifecycle.godsReadyChannel,
timeoutMs: this.maestroConfig.readyTimeoutMs,
matchesChan: this.arenaCnx.matchesChan.bind(this.arenaCnx),
debug: this.debug,
})
}
wireSystemConnexion(cnx) {
cnx.maestroSrv = this
cnx.accessRights = this.accessRights
cnx.reloadAccessRights = () => this.reloadAccessRights()
cnx.getAccessRights = () => this.getAccessRights()
if(!this.systemCnx || cnx.redisConfig.role === 'primary') {
this.systemCnx = cnx
}
}
wireArenaConnexion(cnx) {
cnx.maestroSrv = this
this.arenaCnxs.push(cnx)
if(!this.arenaCnx || cnx.redisConfig.role === 'primary') {
this.arenaCnx = cnx
this.refreshArenaGroom()
this.refreshPrepareQuorum()
}
}
resetOrchestration() {
this.orchestrationState = MaestroState.IDLE
this.simulationId = null
this.agentIds = []
this.keyframeId = null
this.infraId = null
this.bigBangEpoch = null
this.resumeEpoch = null
this.pausedAt = null
this.prepareQuorum?.cancel()
}
handlePrepareAck(msg, chan) {
if(this.orchestrationState !== MaestroState.PREPARING) return(false)
if(!this.prepareQuorum) return(false)
return(this.prepareQuorum.handleMessage(msg, chan))
}
async publishLifecycle(eventType, payload, state = null) {
if(!this.arenaCnx) throw(new Error('No arena Redis connection'))
const resolvedState = state ?? this.orchestrationStateFor(payload?.simulationId ?? this.simulationId)
await this.arenaCnx.redisPublish(this.maestroConfig.lifecycle.arenaChannel, {
eventType,
sender: this.maestroConfig.senderId,
payload,
})
await this.publishSystemLifecycle(eventType, payload, resolvedState)
if(this.debug) console.log(`[Maestro] Published ${eventType} simulationId=${payload.simulationId}`)
}
async publishSystemLifecycle(eventType, payload, state) {
if(!this.systemCnx || !this.simRepo) return
const simulationId = payload?.simulationId
if(!simulationId) return
const ownersResult = await this.simRepo.listSimulationOwnerUuids(simulationId)
if(!ownersResult.ok || !ownersResult.ownerUuids.length) return
const channelTemplate = this.maestroConfig.systemLifecycleChannel
const senderId = this.maestroConfig.senderId
const msg = {
eventType,
sender: senderId,
payload: { ...payload, state },
}
for(const uid of ownersResult.ownerUuids) {
const chan = channelTemplate.replace(/\[UID\]/g, uid)
await this.systemCnx.redisPublish(chan, msg)
}
if(this.debug) {
console.log(`[Maestro] Published system ${eventType} state=${state} simulationId=${simulationId}`)
}
}
async startSimulation(userUuid, payload) {
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
if(!this.arenaGroom) return({ ok: false, err: 'No arena Redis connection' })
const simulationUuid = payload?.simulationUuid
if(this.isPaused()) {
return(this.resumeSimulation(userUuid, simulationUuid))
}
if(!this.prepareQuorum) return({ ok: false, err: 'No prepare quorum (arena Redis not wired)' })
if(!this.isIdle()) return({ ok: false, err: 'A simulation is already in progress' })
const infraId = payload?.infraId ?? null
const access = await this.simRepo.validateSimulationAccess(userUuid, simulationUuid)
if(!access.ok) return(access)
const keyframeId = access.sim.sim_root_kf_uuid
const agentsResult = await this.simRepo.loadKeyframeAgents(keyframeId)
if(!agentsResult.ok) return(agentsResult)
await this.arenaGroom.clearArena()
await this.arenaGroom.seedAgents(agentsResult.agents)
this.simulationId = simulationUuid
this.agentIds = agentsResult.agents.map(a => a.id)
this.keyframeId = keyframeId
this.infraId = infraId
this.orchestrationState = MaestroState.PREPARING
const lifecyclePayload = {
simulationId: this.simulationId,
agentIds: this.agentIds,
keyframeId,
infraId,
}
const expectedParticipants = buildPrepareQuorum(this.agentIds, this.rootConfig)
const readyWait = this.prepareQuorum.begin(expectedParticipants, this.simulationId)
await this.publishLifecycle('onYourMarks', lifecyclePayload)
this.continueStartSimulation(simulationUuid, readyWait).catch(err => {
console.error('[Maestro] continueStartSimulation failed:', err)
})
return({
ok: true,
simulationId: this.simulationId,
keyframeId,
infraId,
agentIds: this.agentIds,
resumed: false,
state: MaestroState.PREPARING,
})
}
async continueStartSimulation(simulationUuid, readyWait) {
try {
const quorum = await readyWait
if(!quorum.ok) {
if(this.orchestrationState === MaestroState.PREPARING && this.simulationId === simulationUuid) {
await this.publishSystemLifecycle('simulationPrepareFailed', {
simulationId: simulationUuid,
err: quorum.err,
}, MaestroState.IDLE)
await this.arenaGroom.clearArena()
this.resetOrchestration()
}
if(this.debug) {
console.warn(`[Maestro] Prepare failed simulationId=${simulationUuid}: ${quorum.err}`)
}
return
}
if(this.orchestrationState !== MaestroState.PREPARING || this.simulationId !== simulationUuid) {
return
}
this.bigBangEpoch = performance.now()
this.resumeEpoch = null
this.pausedAt = null
this.orchestrationState = MaestroState.LIVE
await this.publishLifecycle('bigBang', {
simulationId: this.simulationId,
agentIds: this.agentIds,
})
if(this.debug) console.log(`[Maestro] LIVE simulationId=${this.simulationId}`)
} catch(err) {
console.error(`[Maestro] continueStartSimulation error simulationId=${simulationUuid}:`, err)
if(this.simulationId === simulationUuid && this.orchestrationState === MaestroState.PREPARING) {
await this.publishSystemLifecycle('simulationPrepareFailed', {
simulationId: simulationUuid,
err: err.message ?? 'Prepare failed',
}, MaestroState.IDLE)
await this.arenaGroom?.clearArena()
this.resetOrchestration()
}
}
}
async resumeSimulation(userUuid, simulationUuid) {
if(this.simulationId !== simulationUuid) {
return({ ok: false, err: 'Another simulation is paused' })
}
const access = await this.simRepo.validateSimulationAccess(userUuid, simulationUuid)
if(!access.ok) return(access)
await this.publishLifecycle('simulationResumed', {
simulationId: simulationUuid,
agentIds: [...this.agentIds],
t: this.pausedAt,
})
this.resumeEpoch = performance.now()
this.orchestrationState = MaestroState.LIVE
if(this.debug) console.log(`[Maestro] RESUMED at t=${this.pausedAt}, simulationId=${simulationUuid}`)
return({
ok: true,
simulationId: this.simulationId,
keyframeId: this.keyframeId,
infraId: this.infraId,
agentIds: [...this.agentIds],
resumed: true,
state: MaestroState.LIVE,
})
}
async pauseSimulation(userUuid, payload) {
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
const simulationUuid = payload?.simulationUuid
const access = await this.simRepo.validateSimulationAccess(userUuid, simulationUuid)
if(!access.ok) return(access)
if(this.simulationId !== simulationUuid) {
if(this.simulationId) return({ ok: false, err: 'Another simulation is active' })
return({ ok: false, err: 'Simulation is not running' })
}
if(this.orchestrationState === MaestroState.PAUSED) {
return({ ok: true, simulationId: simulationUuid, t: this.pausedAt })
}
if(this.orchestrationState !== MaestroState.LIVE && this.orchestrationState !== MaestroState.PREPARING) {
return({ ok: false, err: 'Simulation is not running' })
}
const t = this.orchestrationState === MaestroState.LIVE ? this.simNow() : 0
await this.publishLifecycle('simulationPaused', {
simulationId: simulationUuid,
agentIds: [...this.agentIds],
t,
})
this.pausedAt = t
this.orchestrationState = MaestroState.PAUSED
if(this.debug) console.log(`[Maestro] PAUSED at t=${t}, simulationId=${simulationUuid}`)
return({ ok: true, simulationId: simulationUuid, t })
}
async stopSimulation(userUuid, payload) {
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
if(!this.arenaGroom) return({ ok: false, err: 'No arena Redis connection' })
const simulationUuid = payload?.simulationUuid
const access = await this.simRepo.validateSimulationOwner(userUuid, simulationUuid)
if(!access.ok) return(access)
if(this.simulationId && this.simulationId !== simulationUuid) {
return({ ok: false, err: 'Another simulation is active' })
}
if(this.simulationId === simulationUuid) {
let t = null
if(this.orchestrationState === MaestroState.LIVE) {
t = this.simNow()
} else if(this.orchestrationState === MaestroState.PREPARING) {
t = 0
} else if(this.orchestrationState === MaestroState.PAUSED) {
t = this.pausedAt
}
await this.publishLifecycle('simulationStopped', {
simulationId: simulationUuid,
agentIds: [...this.agentIds],
t,
}, MaestroState.IDLE)
}
if(this.simulationId === simulationUuid && this.orchestrationState === MaestroState.PREPARING) {
this.prepareQuorum?.abortPending({ ok: false, err: 'Simulation stopped' })
}
await this.arenaGroom.clearArena()
this.resetOrchestration()
if(this.debug) console.log(`[Maestro] Stopped simulationId=${simulationUuid}`)
return({ ok: true, simulationId: simulationUuid })
}
orchestrationStateFor(simulationId) {
if(this.simulationId !== simulationId) return(MaestroState.IDLE)
return(this.orchestrationState)
}
async getSimulationsStatus(userUuid) {
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
const listResult = await this.simRepo.listOwnerSimulations(userUuid)
if(!listResult.ok) return(listResult)
const simulations = listResult.simulationIds.map(simulationId => ({
simulationId,
state: this.orchestrationStateFor(simulationId),
}))
return({ ok: true, simulations })
}
async reloadAccessRights() {
await this.configHelper.refreshAccessRights()
this.rootConfig.accessRights = this.configHelper.config.accessRights
this.accessRights.refreshAccessRights(this.rootConfig)
}
getAccessRights() {
return(this.rootConfig.accessRights)
}
}
@@ -2,4 +2,5 @@ export const MaestroState = {
IDLE: 'idle',
PREPARING: 'preparing',
LIVE: 'live',
PAUSED: 'paused',
}
@@ -3,10 +3,11 @@ import yargs from 'yargs/yargs'
import { hideBin } from 'yargs/helpers'
import 'node:process'
import { RedisConnexion } from '../redisConnexion.js'
import { busReplyRoute } from '../bus/publishActionReply.js'
import { configHelper } from '../configHelper.js'
import { maestroServer } from './maestroServer.js'
import * as systemMesh from './actions/system/index.js'
import * as arenaMesh from './actions/arena/index.js'
import * as systemMesh from './handlers/system/index.js'
import * as arenaMesh from './handlers/arena/index.js'
const meshModules = {
system: systemMesh,
@@ -25,7 +26,7 @@ console.log = (...args) => logWithTimestamp(originalLog, 'LOG', ...args)
console.warn = (...args) => logWithTimestamp(originalWarn, 'WARN', ...args)
console.error = (...args) => logWithTimestamp(originalError, 'ERROR', ...args)
const argv = yargs(hideBin(process.argv)).command('SimMaestro', 'Simulation orchestrator for P42', {})
const argv = yargs(hideBin(process.argv)).command('Maestro', 'Simulation orchestrator for P42', {})
.options({
'debug': {
description: 'shows debug info',
@@ -49,6 +50,7 @@ let cfgh = new configHelper({
function meshRedisConns(mesh, meshName, debug, rootConfig) {
const { redis, ...meshConfig } = mesh
const busRoute = busReplyRoute(rootConfig.maestro, meshName)
return redis.map(cfg =>
new RedisConnexion({
debug,
@@ -56,6 +58,8 @@ function meshRedisConns(mesh, meshName, debug, rootConfig) {
redisId: cfg.redisId,
meshName,
meshModule: meshModules[meshName],
senderId: busRoute?.senderId,
actionsReply: busRoute?.actionsReply,
})
)
}
@@ -1,5 +1,5 @@
{
"name": "p42SimMaestro",
"name": "p42Maestro",
"version": "1.0.0",
"description": "Simulation orchestrator God-daemon for P42",
"type": "module",
+115
View File
@@ -0,0 +1,115 @@
export class PrepareQuorum {
constructor({ ackChannel, timeoutMs, matchesChan, debug = false }) {
this.ackChannel = ackChannel
this.timeoutMs = timeoutMs
this.matchesChan = matchesChan
this.debug = debug
this.expected = new Set()
this.ready = new Map()
this.simulationId = null
this.active = false
this.resolve = null
this.timer = null
}
begin(expectedParticipantIds, simulationId) {
this.cancel()
this.expected = new Set(expectedParticipantIds)
this.ready.clear()
this.simulationId = simulationId
this.active = true
if(this.debug) {
console.log(
`[Maestro] Prepare quorum armed: ${this.expected.size} participant(s) ` +
`(agents + primordial daemons), timeout ${this.timeoutMs}ms`
)
}
return(new Promise(resolve => {
this.resolve = resolve
this.timer = setTimeout(() => {
const missing = [...this.expected].filter(id => !this.ready.has(id))
this.#finish({
ok: false,
err: `Timeout waiting for readyToStart (${this.timeoutMs}ms); ` +
`missing: ${missing.join(', ') || 'unknown'}`,
})
}, this.timeoutMs)
}))
}
handleMessage(msg, chan) {
if(!this.active) return(false)
if(typeof(this.matchesChan) !== 'function') return(false)
if(!this.matchesChan(chan, this.ackChannel)) return(false)
if(msg?.eventType !== 'readyToStart') return(false)
const payload = msg.payload ?? {}
if(payload.simulationId !== this.simulationId) return(false)
const sender = msg.sender
if(!sender || !this.expected.has(sender)) {
if(this.debug) {
console.warn(`[Maestro] Ignoring readyToStart from unexpected participant: ${sender}`)
}
return(true)
}
if(!payload.success) {
this.#finish({
ok: false,
err: payload.err ?? `Participant ${sender} failed prepare`,
})
return(true)
}
this.ready.set(sender, payload)
if(this.debug) {
console.log(
`[Maestro] readyToStart from ${sender} ` +
`(${this.ready.size}/${this.expected.size})`
)
}
for(const participantId of this.expected) {
if(!this.ready.has(participantId)) return(true)
}
this.#finish({ ok: true })
return(true)
}
cancel() {
this.#cleanup()
}
abortPending(result) {
const resolve = this.resolve
this.#cleanup()
if(typeof(resolve) !== 'function') return(false)
resolve(result)
return(true)
}
#cleanup() {
if(this.timer) {
clearTimeout(this.timer)
this.timer = null
}
this.active = false
this.resolve = null
this.expected.clear()
this.ready.clear()
this.simulationId = null
}
#finish(result) {
const resolve = this.resolve
this.#cleanup()
if(typeof(resolve) === 'function') resolve(result)
}
}
+23
View File
@@ -0,0 +1,23 @@
const SKIP_PRIMORDIAL_SECTIONS = new Set([
'maestro',
'mysql',
'accessRights',
'systemMesh',
'arenaMesh',
])
export function getPrimordialDaemonIds(config = {}) {
const ids = []
for(const [section, block] of Object.entries(config)) {
if(SKIP_PRIMORDIAL_SECTIONS.has(section)) continue
if(!block || typeof(block) !== 'object' || Array.isArray(block)) continue
if(!block.primordialDaemon) continue
ids.push(typeof(block.senderId) === 'string' ? block.senderId : section)
}
return(ids)
}
export function buildPrepareQuorum(agentIds, config) {
const primordialIds = getPrimordialDaemonIds(config)
return([...agentIds, ...primordialIds])
}
+174
View File
@@ -0,0 +1,174 @@
import { MySQLClient } from '@p42/p42modules'
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
export function isValidUuid(val) {
return(typeof(val) === 'string' && UUID_RE.test(val))
}
export class SimRepository {
constructor(dbPool, databases, debug = false) {
this.db = dbPool
this.guiDb = databases.guiDatabase
this.simDb = databases.simDatabase
this.debug = debug
}
#qualify(db, table) {
return(`\`${db}\`.${table}`)
}
#parseJsonObject(raw) {
let v = raw
if(v == null) return(null)
if(typeof(v) === 'string') {
try { v = JSON.parse(v) }
catch { return(null) }
}
if(typeof(v) !== 'object' || Array.isArray(v)) return(null)
return(v)
}
#parseGpsValues(raw) {
const v = this.#parseJsonObject(raw)
if(!v) return(null)
const position = v.position
const speed = v.speed ?? v.vector
if(!position || !speed) return(null)
const axes = ['x', 'y', 'z']
for(const axis of axes) {
if(typeof(position[axis]) !== 'number' || !Number.isFinite(position[axis])) return(null)
if(typeof(speed[axis]) !== 'number' || !Number.isFinite(speed[axis])) return(null)
}
return({
position: { x: position.x, y: position.y, z: position.z },
vector: { x: speed.x, y: speed.y, z: speed.z },
})
}
#parseStoreValues(raw) {
const v = this.#parseJsonObject(raw)
if(v == null) return(null)
return({ ...v })
}
async validateSimulationAccess(userUuid, simulationUuid) {
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
const rows = await MySQLClient.poolExecute(this.db, `
SELECT sim_id,
BIN_TO_UUID(sim_uuid) AS sim_uuid,
BIN_TO_UUID(sim_root_kf_uuid) AS sim_root_kf_uuid
FROM ${this.#qualify(this.simDb, 'simulations')}
INNER JOIN ${this.#qualify(this.guiDb, 'simowners')} ON own_sim_uuid = sim_uuid
INNER JOIN ${this.#qualify(this.guiDb, 'users')} ON own_usr_id = usr_id
WHERE usr_uuid = ?
AND sim_uuid = UUID_TO_BIN(?)
`, [userUuid, simulationUuid])
if(!rows.length) return({ ok: false, err: 'Simulation not found or access denied' })
const sim = rows[0]
if(!sim.sim_root_kf_uuid) return({ ok: false, err: 'Simulation has no root keyframe' })
const kfRows = await MySQLClient.poolExecute(this.db, `
SELECT ekf_uuid
FROM ${this.#qualify(this.simDb, 'edited_keyframes')}
WHERE ekf_uuid = UUID_TO_BIN(?)
`, [sim.sim_root_kf_uuid])
if(!kfRows.length) return({ ok: false, err: 'Root keyframe not found' })
return({ ok: true, sim })
}
async validateSimulationOwner(userUuid, simulationUuid) {
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
const rows = await MySQLClient.poolExecute(this.db, `
SELECT sim_id,
BIN_TO_UUID(sim_uuid) AS sim_uuid,
BIN_TO_UUID(sim_root_kf_uuid) AS sim_root_kf_uuid
FROM ${this.#qualify(this.simDb, 'simulations')}
INNER JOIN ${this.#qualify(this.guiDb, 'simowners')} ON own_sim_uuid = sim_uuid
INNER JOIN ${this.#qualify(this.guiDb, 'users')} ON own_usr_id = usr_id
WHERE usr_uuid = ?
AND sim_uuid = UUID_TO_BIN(?)
`, [userUuid, simulationUuid])
if(!rows.length) return({ ok: false, err: 'Simulation not found or access denied' })
return({ ok: true, sim: rows[0] })
}
async listSimulationOwnerUuids(simulationUuid) {
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
const rows = await MySQLClient.poolExecute(this.db, `
SELECT BIN_TO_UUID(usr_uuid) AS owner_uuid
FROM ${this.#qualify(this.guiDb, 'simowners')}
INNER JOIN ${this.#qualify(this.guiDb, 'users')} ON own_usr_id = usr_id
WHERE own_sim_uuid = UUID_TO_BIN(?)
`, [simulationUuid])
return({
ok: true,
ownerUuids: rows.map(row => row.owner_uuid),
})
}
async listOwnerSimulations(userUuid) {
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
const rows = await MySQLClient.poolExecute(this.db, `
SELECT BIN_TO_UUID(sim_uuid) AS simulation_id
FROM ${this.#qualify(this.simDb, 'simulations')}
INNER JOIN ${this.#qualify(this.guiDb, 'simowners')} ON own_sim_uuid = sim_uuid
INNER JOIN ${this.#qualify(this.guiDb, 'users')} ON own_usr_id = usr_id
WHERE usr_uuid = ?
ORDER BY sim_id
`, [userUuid])
return({
ok: true,
simulationIds: rows.map(row => row.simulation_id),
})
}
async loadKeyframeAgents(keyframeId) {
const rows = await MySQLClient.poolExecute(this.db, `
SELECT BIN_TO_UUID(ekfs_agent_id) AS agent_id,
ekfs_gps_values,
ekfs_store_values
FROM ${this.#qualify(this.simDb, 'edited_kf_store')}
WHERE ekfs_ekf_uuid = UUID_TO_BIN(?)
`, [keyframeId])
const agents = []
const errors = []
for(const row of rows) {
const parsed = this.#parseGpsValues(row.ekfs_gps_values)
if(!parsed) {
errors.push(`Invalid GPS values for agent ${row.agent_id}`)
continue
}
const store = this.#parseStoreValues(row.ekfs_store_values)
if(store == null) {
errors.push(`Invalid store values for agent ${row.agent_id}`)
continue
}
agents.push({
id: row.agent_id,
position: parsed.position,
vector: parsed.vector,
store,
})
}
if(errors.length) return({ ok: false, err: errors.join('; '), agents: [] })
return({ ok: true, agents })
}
}
+45
View File
@@ -0,0 +1,45 @@
#!/bin/sh
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
# shellcheck source=../lib/resolveConfigPath.sh
. "$SCRIPT_DIR/../lib/resolveConfigPath.sh"
CONFIG="$(resolveConfigPath "$SCRIPT_DIR" "${1:-${CONFIG:-../config.json}}")"
set -a
. /etc/p42/secrets.env
set +a
daemon=p42Maestro
logfile=maestro.log
if [ ! -f "$CONFIG" ]; then
echo "Config file not found: $CONFIG" >&2
exit 1
fi
pid=$(pgrep -f "$daemon")
if [ -z "$pid" ]
then
node "${daemon}.js" --config "$CONFIG" --debug > "$logfile" 2>&1 &
pid=$!
sleep 1
if kill -0 "$pid" 2>/dev/null
then
echo ""
echo "$daemon is now running with PID=$pid (config=$CONFIG)"
echo ""
else
echo ""
echo "Failed to start $daemon. Check maestro.log"
echo ""
exit 1
fi
else
echo ""
echo "$daemon is already running with PID=$pid"
echo ""
fi
@@ -1,6 +1,6 @@
#!/bin/sh
pid=`ps -ef | grep p42SimMaestro.js |grep -v grep | awk '{print $2}'`
pid=`ps -ef | grep p42Maestro.js |grep -v grep | awk '{print $2}'`
if [ -n "$pid" ]
then
echo "killing pid: $pid"
-12
View File
@@ -1,12 +0,0 @@
export const construct = (redisCnx) => {
}
export const methods = {
dispatchArenaMessage(msg, chan) {
if(this.debug) console.log(`[${this.redisId}] Arena message (unhandled):`, msg.eventType, chan)
return(false)
},
}
-5
View File
@@ -1,5 +0,0 @@
export function dispatchMessage(redisCnx, msg, chan) {
if(typeof(redisCnx.dispatchArenaMessage) !== 'function') return
redisCnx.dispatchArenaMessage(msg, chan)
}
-12
View File
@@ -1,12 +0,0 @@
import { methods as arenaMethods, construct as arenaConstruct } from './arenaHandlers.js'
import { dispatchMessage } from './dispatch.js'
export const afterLoginMethods = [
arenaConstruct,
]
export const meshActions = {
...arenaMethods,
}
export { dispatchMessage }
-28
View File
@@ -1,28 +0,0 @@
export function dispatchMessage(redisCnx, msg, chan) {
const observer = redisCnx.config.observer
if(!observer?.observerActionsChannel) return
const actionsChan = redisCnx.fullChan(observer.observerActionsChannel)
if(chan != actionsChan) return
const action = msg.action
if(!action || typeof(action) !== 'string') {
console.warn(`[${redisCnx.redisId}] Ignoring message without action on ${chan}`)
return
}
const handler = redisCnx['action_'+action]
if(typeof(handler) != 'function') {
if(redisCnx.debug) console.warn(`[${redisCnx.redisId}] Unknown action ${action} on ${chan}`)
return
}
const payload = ('payload' in msg) ? msg.payload : null
const reqid = ('reqid' in msg) ? msg.reqid.substr(0, 50) : null
const sender = msg.sender || null
const roles = Array.isArray(msg.roles) ? msg.roles : ['*']
if(redisCnx.debug) console.log(`[${redisCnx.redisId}] Dispatching action ${action} from ${sender}`)
handler.call(redisCnx, action, payload, reqid, sender, roles)
}
-12
View File
@@ -1,12 +0,0 @@
import { methods as utilities, construct as utilitiesConstruct } from './utilities.js'
import { dispatchMessage } from './dispatch.js'
export const afterLoginMethods = [
utilitiesConstruct,
]
export const meshActions = {
...utilities,
}
export { dispatchMessage }
-48
View File
@@ -1,48 +0,0 @@
import { publishActionReply } from '../../actionsHelper.js'
export const construct = (redisCnx) => {
}
export const methods = {
async action_RELOADCONFIG(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.observer.observerActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
this.reloadAccessRights()
publishActionReply(this, { ...replyOpts, reply: {
success: true,
} })
},
async action_GETCONFIG(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.observer.observerActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: this.getAccessRights(),
} })
},
}
+4 -14
View File
@@ -1,16 +1,6 @@
export function publishActionReply(redisCnx, options) {
const {
action,
reqid,
sender,
reply,
replyChannel,
senderId = 'observer',
} = options
reply.action = action
reply.sender = senderId
if(reqid) reply.reqid = reqid
const chan = replyChannel.replace(/\[UID\]/g, sender)
redisCnx.redisPublish(chan, reply)
export function parseSimTime(payload, fallbackFn) {
if(payload?.t != null && typeof(payload.t) === 'number' && !Number.isNaN(payload.t)) return(payload.t)
if(payload?.at != null && typeof(payload.at) === 'number' && !Number.isNaN(payload.at)) return(payload.at)
return(fallbackFn())
}
+33
View File
@@ -0,0 +1,33 @@
export class Frustum {
constructor(planes) {
if(!Frustum.#validatePlanes(planes)) throw(new Error('Invalid frustum planes'))
this.planes = planes.map(p => ({ ...p }))
}
static #validatePlanes(planes) {
if(!Array.isArray(planes) || planes.length !== 6) return(false)
for(const p of planes) {
if(!p || typeof(p) !== 'object') return(false)
if(typeof(p.nx) !== 'number' || Number.isNaN(p.nx)) return(false)
if(typeof(p.ny) !== 'number' || Number.isNaN(p.ny)) return(false)
if(typeof(p.nz) !== 'number' || Number.isNaN(p.nz)) return(false)
if(typeof(p.d) !== 'number' || Number.isNaN(p.d)) return(false)
}
return(true)
}
static fromPlanes(planes) {
if(!Frustum.#validatePlanes(planes)) return(null)
return(new Frustum(planes))
}
containsPoint(position) {
for(const p of this.planes) {
const dist = p.nx * position.x + p.ny * position.y + p.nz * position.z + p.d
if(dist < 0) return(false)
}
return(true)
}
}
+82
View File
@@ -0,0 +1,82 @@
import { positionAt } from '../GPS/handlers/arena/worldline.js'
export class GpsStorageReader {
constructor(systemCnx, gpsStorage, debug = false) {
this.cnx = systemCnx
this.gpsStorage = gpsStorage
this.debug = debug
}
agentHashKey(agentId) {
return(this.gpsStorage.agentHashKey.replace(/\[UID\]/g, agentId))
}
#parseHashField(raw) {
if(raw == null) return(null)
if(typeof(raw) === 'object') return(raw)
try { return(JSON.parse(raw)) }
catch { return(null) }
}
#segmentToAgent(segment) {
if(!segment?.id) return(null)
if(!this.#isVector(segment.position) || !this.#isVector(segment.vector)) return(null)
return({
id: segment.id,
position: { ...segment.position },
vector: { ...segment.vector },
since: segment.since ?? 0,
generation: segment.generation ?? 0,
})
}
#isVector(v) {
return(
v &&
typeof(v) === 'object' &&
typeof(v.x) === 'number' &&
typeof(v.y) === 'number' &&
typeof(v.z) === 'number'
)
}
buildAgentSnapshot(agent, at) {
return({
id: agent.id,
position: positionAt(agent, at),
vector: { ...agent.vector },
since: agent.since,
generation: agent.generation ?? 0,
t: at,
})
}
async loadSegment(agentId) {
const raw = await this.cnx.redisHget(this.agentHashKey(agentId), 'segment')
return(this.#parseHashField(raw))
}
async listAgentIds() {
return(await this.cnx.redisSmembers(this.gpsStorage.agentsIndexKey))
}
async loadAllAgents() {
const ids = await this.listAgentIds()
const agents = new Map()
for(const id of ids) {
const segment = await this.loadSegment(id)
const agent = this.#segmentToAgent(segment)
if(agent) agents.set(id, agent)
}
return(agents)
}
async getAgentPosition(agentId, at) {
const segment = await this.loadSegment(agentId)
const agent = this.#segmentToAgent(segment)
if(!agent) return(null)
return(this.buildAgentSnapshot(agent, at))
}
}
+17
View File
@@ -0,0 +1,17 @@
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
import * as lifecycle from './lifecycle.js'
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([lifecycle])
export { actionHandlers, afterLogin }
export const dispatchMessage = createDispatchMessage({
eventHandlers,
actionRules(redisCnx) {
const observer = redisCnx.config.observer ?? {}
const arenaChannel = observer.arenaActionsChannel
return({
channels: arenaChannel ? [arenaChannel] : [],
})
},
})
+14
View File
@@ -0,0 +1,14 @@
export const eventHandlers = {
'arena:lifecycle': {
onYourMarks(msg, chan, sender, cnxId) {
this.observerSrv?.onYourMarks()
},
bigBang(msg, chan, sender, cnxId) {
this.observerSrv?.onBigBang()
},
simulationStopped(msg, chan, sender, cnxId) {
this.observerSrv?.onSimulationStopped(msg.payload ?? {})
},
},
}
+17
View File
@@ -0,0 +1,17 @@
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
import * as positions from './positions.js'
import * as utilities from './utilities.js'
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([positions, utilities])
export { actionHandlers, afterLogin }
export const dispatchMessage = createDispatchMessage({
eventHandlers,
actionRules(redisCnx) {
const observer = redisCnx.config.observer ?? {}
return({
channels: [observer.ActionsChannel].filter(Boolean),
})
},
})
+122
View File
@@ -0,0 +1,122 @@
import { replyToAction } from '../../../bus/publishActionReply.js'
import { parseSimTime } from '../../actionsHelper.js'
import { Frustum } from '../../frustum.js'
export const actions = {
async action_GETAGENTPOSITION(action, payload, reqid, sender, cnxId, roles) {
const reader = this.observerSrv.gpsStorageReader
if(!reader) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'GPS storage reader not ready' })
return
}
if(!this.observerSrv.isLive()) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Simulation not live' })
return
}
const agentId = payload?.agentId
if(!agentId || typeof(agentId) !== 'string') {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid agentId' })
return
}
const at = parseSimTime(payload, () => this.observerSrv.now())
if(at === null) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Invalid simulation time' })
return
}
const agent = await reader.getAgentPosition(agentId, at)
if(!agent) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: `Unknown agent: ${agentId}` })
return
}
replyToAction(this, { action, reqid, sender, cnxId, success: true, payload: { agent } })
},
async action_GETAGENTSINFRUSTUM(action, payload, reqid, sender, cnxId, roles) {
const registry = this.observerSrv.requestorRegistry
if(!registry) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Requestor registry not ready' })
return
}
if(!this.observerSrv.isLive()) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Simulation not live' })
return
}
const frustum = Frustum.fromPlanes(payload?.planes)
if(!frustum) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid frustum planes (expected 6)' })
return
}
const at = parseSimTime(payload, () => this.observerSrv.now())
if(at === null) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Invalid simulation time' })
return
}
const result = await registry.evaluateOnce({ frustum, t: at })
if(!result.ok) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: result.err })
return
}
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: {
agents: result.agents,
t: result.t,
},
})
},
async action_SUBSCRIBEFRUSTUM(action, payload, reqid, sender, cnxId, roles) {
const registry = this.observerSrv.requestorRegistry
if(!registry) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Requestor registry not ready' })
return
}
if(!this.observerSrv.isLive()) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Simulation not live' })
return
}
if(!cnxId || typeof(cnxId) !== 'string') {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: 'Missing or invalid cnxId' })
return
}
const result = await registry.subscribeFrustum(cnxId, {
planes: payload?.planes,
frequency: payload?.frequency,
})
if(!result.ok) {
replyToAction(this, { action, reqid, sender, cnxId, success: false, err: result.err })
return
}
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: {
frequency: result.frequency,
t: result.t,
},
})
},
}
+21
View File
@@ -0,0 +1,21 @@
import { replyToAction } from '../../../bus/publishActionReply.js'
export const actions = {
async action_RELOADCONFIG(action, payload, reqid, sender, cnxId, roles) {
this.reloadAccessRights()
replyToAction(this, { action, reqid, sender, cnxId, success: true })
},
async action_GETCONFIG(action, payload, reqid, sender, cnxId, roles) {
replyToAction(this, {
action,
reqid,
sender,
cnxId,
success: true,
payload: this.getAccessRights(),
})
},
}
+99 -12
View File
@@ -1,29 +1,115 @@
import { AccesRights } from '../accesRights.js'
import { GpsStorageReader } from './gpsStorageReader.js'
import { RequestorRegistry } from './requestorRegistry.js'
import { SimState } from '../GPS/simulationState.js'
export class observerServer {
constructor(configHelper, allRediscnx, debug) {
this.configHelper = configHelper
this.observerConfig = configHelper.config
this.rootConfig = configHelper.config
this.observerConfig = configHelper.config.observer ?? {}
this.gpsConfig = configHelper.config.gps ?? {}
this.allRediscnx = allRediscnx
this.debug = debug
this.accessRights = new AccesRights(this.observerConfig, debug)
this.accessRights = new AccesRights(this.rootConfig, debug)
this.arenaCnx = null
this.arenaCnxs = []
this.systemCnx = null
this.gpsStorageReader = null
this.requestorRegistry = null
this.state = SimState.IDLE
this.bigBangEpoch = null
}
getObserverSettings() {
const observer = this.observerConfig.observer ?? {}
return({
senderId: observer.senderId ?? 'observer',
lifecycle: {
arenaChannel: observer.lifecycle?.arenaChannel ?? 'arena:lifecycle',
godsReadyChannel: observer.lifecycle?.godsReadyChannel ?? 'arena:gods:ready',
initGpsStorageReader() {
const gpsStorage = this.gpsConfig.GPSstorage
if(gpsStorage && this.systemCnx) {
this.gpsStorageReader = new GpsStorageReader(this.systemCnx, gpsStorage, this.debug)
this.initRequestorRegistry()
}
}
initRequestorRegistry() {
if(!this.gpsStorageReader || this.requestorRegistry) return
const { scanIntervalMs } = this.observerConfig
this.requestorRegistry = new RequestorRegistry(
this.gpsStorageReader,
() => this.now(),
scanIntervalMs,
(subscriberId, payload) => this.publishFrustumAgentEvents(
subscriberId,
payload.agents,
payload.t
),
this.debug
)
}
async publishFrustumAgentEvents(subscriberId, agents, t) {
if(!this.systemCnx || !subscriberId) return
if(!Array.isArray(agents) || !agents.length) return
const chan = this.observerConfig.FrustumEventsChannel
.replace(/\[CUID\]/g, subscriberId)
const senderId = this.observerConfig.senderId
for(const agent of agents) {
if(!agent?.id || !agent?.position) continue
await this.systemCnx.redisPublish(chan, {
eventType: 'move',
sender: senderId,
cnxId: this.systemCnx.cnxId,
payload: {
aid: agent.id,
coords: {
x: agent.position.x,
y: agent.position.y,
z: agent.position.z,
},
t,
},
})
}
if(this.debug) {
console.log(`[Observer] Frustum events: ${agents.length} agent(s) on ${chan} at t=${t}`)
}
}
isLive() {
return(this.state === SimState.LIVE)
}
simNow() {
if(this.bigBangEpoch === null) return(null)
return((performance.now() - this.bigBangEpoch) / 1000)
}
now() {
if(this.isLive()) return(this.simNow())
return(null)
}
onYourMarks() {
this.state = SimState.PREPARE
this.bigBangEpoch = null
this.requestorRegistry?.clear()
}
onBigBang() {
this.bigBangEpoch = performance.now()
this.state = SimState.LIVE
}
onSimulationStopped(payload = {}) {
this.requestorRegistry?.clear()
this.state = SimState.IDLE
this.bigBangEpoch = null
if(this.debug) console.log(`[Observer] simulationStopped at t=${payload.t}`)
}
wireSystemConnexion(cnx) {
cnx.observerSrv = this
cnx.accessRights = this.accessRights
@@ -31,6 +117,7 @@ export class observerServer {
cnx.getAccessRights = () => this.getAccessRights()
if(!this.systemCnx || cnx.redisConfig.role === 'primary') {
this.systemCnx = cnx
this.initGpsStorageReader()
}
}
@@ -44,12 +131,12 @@ export class observerServer {
async reloadAccessRights() {
await this.configHelper.refreshAccessRights()
this.observerConfig.accessRights = this.configHelper.config.accessRights
this.accessRights.refreshAccessRights(this.observerConfig)
this.rootConfig.accessRights = this.configHelper.config.accessRights
this.accessRights.refreshAccessRights(this.rootConfig)
}
getAccessRights() {
return(this.observerConfig.accessRights)
return(this.rootConfig.accessRights)
}
}
+6 -2
View File
@@ -3,10 +3,11 @@ import yargs from 'yargs/yargs'
import { hideBin } from 'yargs/helpers'
import 'node:process'
import { RedisConnexion } from '../redisConnexion.js'
import { busReplyRoute } from '../bus/publishActionReply.js'
import { configHelper } from '../configHelper.js'
import { observerServer } from './observerServer.js'
import * as systemMesh from './actions/system/index.js'
import * as arenaMesh from './actions/arena/index.js'
import * as systemMesh from './handlers/system/index.js'
import * as arenaMesh from './handlers/arena/index.js'
const meshModules = {
system: systemMesh,
@@ -49,6 +50,7 @@ let cfgh = new configHelper({
function meshRedisConns(mesh, meshName, debug, rootConfig) {
const { redis, ...meshConfig } = mesh
const busRoute = busReplyRoute(rootConfig.observer, meshName)
return redis.map(cfg =>
new RedisConnexion({
debug,
@@ -56,6 +58,8 @@ function meshRedisConns(mesh, meshName, debug, rootConfig) {
redisId: cfg.redisId,
meshName,
meshModule: meshModules[meshName],
senderId: busRoute?.senderId,
actionsReply: busRoute?.actionsReply,
})
)
}
+187
View File
@@ -0,0 +1,187 @@
import { positionAt } from '../GPS/handlers/arena/worldline.js'
import { Frustum } from './frustum.js'
export class RequestorRegistry {
constructor(reader, getNow, scanIntervalMs, onPush, debug = false) {
this.reader = reader
this.getNow = getNow
this.scanIntervalMs = scanIntervalMs
this.onPush = onPush
this.debug = debug
this.requestors = new Map()
}
#tickTimer = null
#scanInFlight = false
#adjustFrequency(requestedMs) {
if(typeof(requestedMs) !== 'number' || Number.isNaN(requestedMs)) return(null)
if(requestedMs < 300 || requestedMs > 10000) return(null)
const tickMs = this.scanIntervalMs
const ticks = Math.round(requestedMs / tickMs)
const minTicks = Math.ceil(300 / tickMs)
const maxTicks = Math.floor(10000 / tickMs)
const clampedTicks = Math.max(minTicks, Math.min(maxTicks, ticks))
return(clampedTicks * tickMs)
}
async subscribeFrustum(id, { planes, frequency }) {
if(!id || typeof(id) !== 'string') return({ ok: false, err: 'Invalid requestor id' })
const frustum = Frustum.fromPlanes(planes)
if(!frustum) return({ ok: false, err: 'Invalid frustum planes' })
const frequencyMs = this.#adjustFrequency(frequency)
if(frequencyMs === null) {
return({ ok: false, err: 'Invalid frequency (expected 30010000 ms)' })
}
const t = this.getNow()
if(t === null) return({ ok: false, err: 'Simulation not live' })
const agents = await this.reader.loadAllAgents()
const matching = this.#matchAgents(agents, frustum, t)
const pushEveryNTicks = frequencyMs / this.scanIntervalMs
this.requestors.set(id, {
id,
frustum,
tMode: 'live',
subscription: true,
frequencyMs,
pushEveryNTicks,
tickCounter: 0,
agents: matching,
t,
updatedAt: Date.now(),
})
this.#ensureTick()
this.#pushAgentEvents(this.requestors.get(id))
return({
ok: true,
frequency: frequencyMs,
t,
})
}
updateRequestor(id, spec) {
const requestor = this.requestors.get(id)
if(!requestor) return({ ok: false, err: 'Unknown requestor' })
if(Array.isArray(spec?.planes)) {
const frustum = Frustum.fromPlanes(spec.planes)
if(!frustum) return({ ok: false, err: 'Invalid frustum planes' })
requestor.frustum = frustum
}
if(typeof(spec?.frequency) === 'number') {
const frequencyMs = this.#adjustFrequency(spec.frequency)
if(frequencyMs === null) return({ ok: false, err: 'Invalid frequency (expected 30010000 ms)' })
requestor.frequencyMs = frequencyMs
requestor.pushEveryNTicks = frequencyMs / this.scanIntervalMs
requestor.tickCounter = 0
}
return({ ok: true, frequency: requestor.frequencyMs })
}
unregisterRequestor(id) {
this.requestors.delete(id)
if(this.requestors.size === 0) this.#stopTick()
return({ ok: true })
}
getRequestorAgents(id) {
const requestor = this.requestors.get(id)
if(!requestor) return(null)
return({
agents: [...requestor.agents],
t: requestor.t,
frequency: requestor.frequencyMs ?? null,
updatedAt: requestor.updatedAt,
})
}
clear() {
this.requestors.clear()
this.#stopTick()
}
async evaluateOnce({ frustum, t }) {
if(!frustum || !(frustum instanceof Frustum)) {
return({ ok: false, err: 'Invalid frustum', agents: [] })
}
if(typeof(t) !== 'number' || Number.isNaN(t)) return({ ok: false, err: 'Invalid simulation time', agents: [] })
const agents = await this.reader.loadAllAgents()
const matching = this.#matchAgents(agents, frustum, t)
return({ ok: true, agents: matching, t })
}
#resolveT(requestor) {
if(requestor.tMode === 'fixed') return(requestor.fixedT)
return(this.getNow())
}
#matchAgents(agents, frustum, t) {
const matching = []
for(const agent of agents.values()) {
const position = positionAt(agent, t)
if(!frustum.containsPoint(position)) continue
matching.push(this.reader.buildAgentSnapshot(agent, t))
}
return(matching)
}
#pushAgentEvents(requestor) {
if(typeof(this.onPush) !== 'function') return
void this.onPush(requestor.id, {
agents: [...requestor.agents],
t: requestor.t,
})
}
async #scanAll() {
if(this.#scanInFlight || this.requestors.size === 0) return
this.#scanInFlight = true
try {
const agents = await this.reader.loadAllAgents()
for(const requestor of this.requestors.values()) {
const t = this.#resolveT(requestor)
if(t === null) {
requestor.agents = []
requestor.t = null
requestor.updatedAt = Date.now()
continue
}
requestor.agents = this.#matchAgents(agents, requestor.frustum, t)
requestor.t = t
requestor.updatedAt = Date.now()
if(!requestor.subscription) continue
requestor.tickCounter++
if(requestor.tickCounter >= requestor.pushEveryNTicks) {
requestor.tickCounter = 0
this.#pushAgentEvents(requestor)
}
}
if(this.debug) console.log(`[Observer] Scanned ${agents.size} agent(s) for ${this.requestors.size} requestor(s)`)
} catch(err) {
console.error('[Observer] Requestor scan failed:', err)
} finally {
this.#scanInFlight = false
}
}
#ensureTick() {
if(this.#tickTimer) return
this.#tickTimer = setInterval(() => {
this.#scanAll()
}, this.scanIntervalMs)
}
#stopTick() {
if(!this.#tickTimer) return
clearInterval(this.#tickTimer)
this.#tickTimer = null
}
}
+16 -2
View File
@@ -1,15 +1,28 @@
#!/bin/sh
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
# shellcheck source=../lib/resolveConfigPath.sh
. "$SCRIPT_DIR/../lib/resolveConfigPath.sh"
CONFIG="$(resolveConfigPath "$SCRIPT_DIR" "${1:-${CONFIG:-../config.json}}")"
set -a
. /etc/p42/secrets.env
set +a
daemon=p42Observer
logfile=observer.log
if [ ! -f "$CONFIG" ]; then
echo "Config file not found: $CONFIG" >&2
exit 1
fi
pid=$(pgrep -f "$daemon")
if [ -z "$pid" ]
then
node "${daemon}.js" --debug > "$logfile" 2>&1 &
node "${daemon}.js" --config "$CONFIG" --debug > "$logfile" 2>&1 &
pid=$!
sleep 1
@@ -17,12 +30,13 @@ then
if kill -0 "$pid" 2>/dev/null
then
echo ""
echo "$daemon is now running with PID=$pid"
echo "$daemon is now running with PID=$pid (config=$CONFIG)"
echo ""
else
echo ""
echo "Failed to start $daemon. Check observer.log"
echo ""
exit 1
fi
else
echo ""
-22
View File
@@ -1,22 +0,0 @@
export const construct = (redisCnx) => {
}
export const methods = {
dispatchArenaMessage(msg, chan) {
const maestro = this.config.maestro
if(!maestro || !this.maestroSrv) return(false)
if(this.matchesChan(chan, maestro.lifecycle?.godsReadyChannel ?? 'arena:gods:ready')) {
if(msg.eventType === 'readyToStart') {
this.maestroSrv.onReadyToStart(msg)
return(true)
}
}
if(this.debug) console.log(`[${this.redisId}] Arena message (unhandled):`, msg.eventType, chan)
return(false)
},
}
-5
View File
@@ -1,5 +0,0 @@
export function dispatchMessage(redisCnx, msg, chan) {
if(typeof(redisCnx.dispatchArenaMessage) !== 'function') return
redisCnx.dispatchArenaMessage(msg, chan)
}
-12
View File
@@ -1,12 +0,0 @@
import { methods as arenaMethods, construct as arenaConstruct } from './arenaHandlers.js'
import { dispatchMessage } from './dispatch.js'
export const afterLoginMethods = [
arenaConstruct,
]
export const meshActions = {
...arenaMethods,
}
export { dispatchMessage }
-28
View File
@@ -1,28 +0,0 @@
export function dispatchMessage(redisCnx, msg, chan) {
const maestro = redisCnx.config.maestro
if(!maestro?.maestroActionsChannel) return
const actionsChan = redisCnx.fullChan(maestro.maestroActionsChannel)
if(chan != actionsChan) return
const action = msg.action
if(!action || typeof(action) !== 'string') {
console.warn(`[${redisCnx.redisId}] Ignoring message without action on ${chan}`)
return
}
const handler = redisCnx['action_'+action]
if(typeof(handler) != 'function') {
if(redisCnx.debug) console.warn(`[${redisCnx.redisId}] Unknown action ${action} on ${chan}`)
return
}
const payload = ('payload' in msg) ? msg.payload : null
const reqid = ('reqid' in msg) ? msg.reqid.substr(0, 50) : null
const sender = msg.sender || null
const roles = Array.isArray(msg.roles) ? msg.roles : ['*']
if(redisCnx.debug) console.log(`[${redisCnx.redisId}] Dispatching action ${action} from ${sender}`)
handler.call(redisCnx, action, payload, reqid, sender, roles)
}
-14
View File
@@ -1,14 +0,0 @@
import { methods as utilities, construct as utilitiesConstruct } from './utilities.js'
import { methods as simulation } from './simulation.js'
import { dispatchMessage } from './dispatch.js'
export const afterLoginMethods = [
utilitiesConstruct,
]
export const meshActions = {
...utilities,
...simulation,
}
export { dispatchMessage }
-138
View File
@@ -1,138 +0,0 @@
import { publishActionReply } from '../../actionsHelper.js'
import { isValidUuid } from '../../simRepository.js'
export const methods = {
/* Event-Rx:
{
"action": "STARTSIMULATION",
"reqid": "6az5e4r6a",
"sender": "<user-uuid>",
"roles": ["*"],
"payload": {
"simulationUuid": "...",
"keyframeId": "...",
"infraId": "..."
}
}
*/
async action_STARTSIMULATION(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.maestro.maestroActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
if(!sender || !isValidUuid(sender)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Missing or invalid sender (user UUID)',
} })
return
}
if(!payload?.simulationUuid || !payload?.keyframeId) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Missing simulationUuid or keyframeId',
} })
return
}
try {
const result = await this.maestroSrv.startSimulation(sender, payload)
if(!result.ok) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: result.err,
} })
return
}
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: {
simulationId: result.simulationId,
keyframeId: result.keyframeId,
infraId: result.infraId,
agentIds: result.agentIds,
},
} })
} catch(err) {
console.error(`[${this.redisId}] STARTSIMULATION failed:`, err)
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: err.message ?? 'STARTSIMULATION failed',
} })
}
},
/* Event-Rx:
{
"action": "STOPSIMULATION",
"reqid": "6az5e4r6a",
"sender": "<user-uuid>",
"payload": { "simulationUuid": "..." }
}
*/
async action_STOPSIMULATION(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.maestro.maestroActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
if(!sender || !isValidUuid(sender)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Missing or invalid sender (user UUID)',
} })
return
}
if(!payload?.simulationUuid) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Missing simulationUuid',
} })
return
}
try {
const result = await this.maestroSrv.stopSimulation(sender, payload)
if(!result.ok) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: result.err,
} })
return
}
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: { simulationId: result.simulationId },
} })
} catch(err) {
console.error(`[${this.redisId}] STOPSIMULATION failed:`, err)
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: err.message ?? 'STOPSIMULATION failed',
} })
}
},
}
-48
View File
@@ -1,48 +0,0 @@
import { publishActionReply } from '../../actionsHelper.js'
export const construct = (redisCnx) => {
}
export const methods = {
async action_RELOADCONFIG(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.maestro.maestroActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
this.reloadAccessRights()
publishActionReply(this, { ...replyOpts, reply: {
success: true,
} })
},
async action_GETCONFIG(action, payload, reqid, sender, roles) {
const replyOpts = {
action,
reqid,
sender,
replyChannel: this.config.maestro.maestroActionsReply,
}
if(!this.accessRights.canDo(roles, action)) {
publishActionReply(this, { ...replyOpts, reply: {
success: false,
err: 'Unauthorized action !',
} })
return
}
publishActionReply(this, { ...replyOpts, reply: {
success: true,
payload: this.getAccessRights(),
} })
},
}
-16
View File
@@ -1,16 +0,0 @@
export function publishActionReply(redisCnx, options) {
const {
action,
reqid,
sender,
reply,
replyChannel,
senderId = 'maestro',
} = options
reply.action = action
reply.sender = senderId
if(reqid) reply.reqid = reqid
const chan = replyChannel.replace(/\[UID\]/g, sender)
redisCnx.redisPublish(chan, reply)
}
-264
View File
@@ -1,264 +0,0 @@
import { AccesRights } from '../accesRights.js'
import { createMysqlPool } from '../mysqlClient.js'
import { SimRepository } from './simRepository.js'
import { ArenaGroom } from './arenaGroom.js'
import { MaestroState } from './orchestrationState.js'
export class maestroServer {
constructor(configHelper, allRediscnx, debug) {
this.configHelper = configHelper
this.maestroConfig = configHelper.config
this.allRediscnx = allRediscnx
this.debug = debug
this.accessRights = new AccesRights(this.maestroConfig, debug)
this.arenaCnx = null
this.arenaCnxs = []
this.systemCnx = null
this.db = null
this.simRepo = null
this.arenaGroom = null
this.orchestrationState = MaestroState.IDLE
this.simulationId = null
this.agentIds = []
this.readyGods = new Map()
this.readyQuorumResolve = null
this.readyQuorumTimer = null
}
getMaestroSettings() {
const maestro = this.maestroConfig.maestro ?? {}
return({
senderId: maestro.senderId ?? 'maestro',
lifecycle: {
arenaChannel: maestro.lifecycle?.arenaChannel ?? 'arena:lifecycle',
godsReadyChannel: maestro.lifecycle?.godsReadyChannel ?? 'arena:gods:ready',
},
expectedGods: maestro.expectedGods ?? ['gps'],
readyTimeoutMs: maestro.readyTimeoutMs ?? 30000,
})
}
getArenaStorageSettings() {
const gps = this.maestroConfig.gps ?? {}
return({
agentHashKey: gps.arenaStorage?.agentHashKey ?? 'arena:agents:[UID]',
agentsIndexKey: gps.arenaStorage?.agentsIndexKey ?? 'arena:agents',
})
}
isIdle() {
return(this.orchestrationState === MaestroState.IDLE)
}
isLive() {
return(this.orchestrationState === MaestroState.LIVE)
}
async init() {
const mysqlCfg = this.maestroConfig.mysql
if(!mysqlCfg) {
console.error('[Maestro] Missing mysql config')
return(false)
}
this.db = await createMysqlPool(mysqlCfg)
this.simRepo = new SimRepository(this.db, this.debug)
if(this.debug) console.log('[Maestro] MySQL pool ready')
return(true)
}
refreshArenaGroom() {
if(this.arenaCnx) {
this.arenaGroom = new ArenaGroom(
this.arenaCnx,
this.getArenaStorageSettings(),
this.debug
)
}
}
wireSystemConnexion(cnx) {
cnx.maestroSrv = this
cnx.accessRights = this.accessRights
cnx.reloadAccessRights = () => this.reloadAccessRights()
cnx.getAccessRights = () => this.getAccessRights()
if(!this.systemCnx || cnx.redisConfig.role === 'primary') {
this.systemCnx = cnx
}
}
wireArenaConnexion(cnx) {
cnx.maestroSrv = this
this.arenaCnxs.push(cnx)
if(!this.arenaCnx || cnx.redisConfig.role === 'primary') {
this.arenaCnx = cnx
this.refreshArenaGroom()
}
}
resetOrchestration() {
this.orchestrationState = MaestroState.IDLE
this.simulationId = null
this.agentIds = []
this.readyGods.clear()
this.clearReadyQuorumWait()
}
clearReadyQuorumWait() {
if(this.readyQuorumTimer) {
clearTimeout(this.readyQuorumTimer)
this.readyQuorumTimer = null
}
this.readyQuorumResolve = null
}
completeReadyQuorum(result) {
const resolve = this.readyQuorumResolve
this.clearReadyQuorumWait()
if(typeof(resolve) === 'function') resolve(result)
}
waitForReadyQuorum() {
const { readyTimeoutMs } = this.getMaestroSettings()
return(new Promise(resolve => {
this.readyQuorumResolve = resolve
this.readyQuorumTimer = setTimeout(() => {
this.completeReadyQuorum({
ok: false,
err: `Timeout waiting for readyToStart (${readyTimeoutMs}ms)`,
})
}, readyTimeoutMs)
}))
}
onReadyToStart(msg) {
if(this.orchestrationState !== MaestroState.PREPARING) return
const payload = msg.payload ?? {}
if(payload.simulationId !== this.simulationId) return
const sender = msg.sender
const { expectedGods } = this.getMaestroSettings()
if(!expectedGods.includes(sender)) {
if(this.debug) console.warn(`[Maestro] Ignoring readyToStart from unexpected sender: ${sender}`)
return
}
if(!payload.success) {
this.completeReadyQuorum({
ok: false,
err: payload.err ?? `Participant ${sender} failed prepare`,
})
return
}
this.readyGods.set(sender, payload)
if(this.debug) console.log(`[Maestro] readyToStart from ${sender} (${this.readyGods.size}/${expectedGods.length})`)
for(const god of expectedGods) {
if(!this.readyGods.has(god)) return
}
this.completeReadyQuorum({ ok: true })
}
async publishLifecycle(eventType, payload) {
if(!this.arenaCnx) throw new Error('No arena Redis connection')
const { arenaChannel, senderId } = this.getMaestroSettings().lifecycle
await this.arenaCnx.redisPublish(arenaChannel, {
eventType,
sender: senderId,
payload,
})
if(this.debug) console.log(`[Maestro] Published ${eventType} simulationId=${payload.simulationId}`)
}
async startSimulation(userUuid, payload) {
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
if(!this.arenaGroom) return({ ok: false, err: 'No arena Redis connection' })
if(!this.isIdle()) return({ ok: false, err: 'A simulation is already in progress' })
const simulationUuid = payload?.simulationUuid
const keyframeId = payload?.keyframeId
const infraId = payload?.infraId ?? null
const access = await this.simRepo.validateSimulationAccess(userUuid, simulationUuid, keyframeId)
if(!access.ok) return(access)
const agentsResult = await this.simRepo.loadKeyframeAgents(keyframeId)
if(!agentsResult.ok) return(agentsResult)
await this.arenaGroom.clearArena()
await this.arenaGroom.seedAgents(agentsResult.agents)
this.simulationId = simulationUuid
this.agentIds = agentsResult.agents.map(a => a.id)
this.readyGods.clear()
this.orchestrationState = MaestroState.PREPARING
const lifecyclePayload = {
simulationId: this.simulationId,
agentIds: this.agentIds,
keyframeId,
infraId,
}
const readyWait = this.waitForReadyQuorum()
await this.publishLifecycle('onYourMarks', lifecyclePayload)
const quorum = await readyWait
if(!quorum.ok) {
await this.arenaGroom.clearArena()
this.resetOrchestration()
return(quorum)
}
await this.publishLifecycle('bigBang', {
simulationId: this.simulationId,
agentIds: this.agentIds,
})
this.orchestrationState = MaestroState.LIVE
if(this.debug) console.log(`[Maestro] LIVE simulationId=${this.simulationId}`)
return({
ok: true,
simulationId: this.simulationId,
keyframeId,
infraId,
agentIds: this.agentIds,
})
}
async stopSimulation(userUuid, payload) {
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
if(!this.arenaGroom) return({ ok: false, err: 'No arena Redis connection' })
const simulationUuid = payload?.simulationUuid
const access = await this.simRepo.validateSimulationOwner(userUuid, simulationUuid)
if(!access.ok) return(access)
if(this.simulationId && this.simulationId !== simulationUuid) {
return({ ok: false, err: 'Another simulation is active' })
}
await this.arenaGroom.clearArena()
this.resetOrchestration()
if(this.debug) console.log(`[Maestro] Stopped simulationId=${simulationUuid}`)
return({ ok: true, simulationId: simulationUuid })
}
async reloadAccessRights() {
await this.configHelper.refreshAccessRights()
this.maestroConfig.accessRights = this.configHelper.config.accessRights
this.accessRights.refreshAccessRights(this.maestroConfig)
}
getAccessRights() {
return(this.maestroConfig.accessRights)
}
}
-116
View File
@@ -1,116 +0,0 @@
import { mysqlExecute } from '../mysqlClient.js'
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
export function isValidUuid(val) {
return(typeof(val) === 'string' && UUID_RE.test(val))
}
function parseGpsValues(raw) {
let v = raw
if(v == null) return(null)
if(typeof(v) === 'string') {
try { v = JSON.parse(v) }
catch { return(null) }
}
if(typeof(v) !== 'object') return(null)
const position = v.position
const speed = v.speed ?? v.vector
if(!position || !speed) return(null)
const axes = ['x', 'y', 'z']
for(const axis of axes) {
if(typeof(position[axis]) !== 'number' || typeof(speed[axis]) !== 'number') return(null)
}
return({
position: { x: position.x, y: position.y, z: position.z },
vector: { x: speed.x, y: speed.y, z: speed.z },
})
}
export class SimRepository {
constructor(dbPool, debug = false) {
this.db = dbPool
this.debug = debug
}
async validateSimulationAccess(userUuid, simulationUuid, keyframeId) {
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
if(!isValidUuid(keyframeId)) return({ ok: false, err: 'Invalid keyframe ID' })
const rows = await mysqlExecute(this.db, `
SELECT s.sim_id,
BIN_TO_UUID(s.sim_uuid) AS sim_uuid,
BIN_TO_UUID(s.sim_root_kf_uuid) AS sim_root_kf_uuid
FROM p42SIM.simulations s
INNER JOIN p42GUI.simowners o ON o.own_sim_uuid = s.sim_uuid
INNER JOIN p42GUI.users u ON o.own_usr_id = u.usr_id
WHERE u.usr_uuid = ?
AND s.sim_uuid = UUID_TO_BIN(?)
`, [userUuid, simulationUuid])
if(!rows.length) return({ ok: false, err: 'Simulation not found or access denied' })
const sim = rows[0]
if(sim.sim_root_kf_uuid !== keyframeId) {
return({ ok: false, err: 'Keyframe does not match simulation root keyframe' })
}
const kfRows = await mysqlExecute(this.db, `
SELECT ekf_uuid
FROM p42SIM.edited_keyframes
WHERE ekf_uuid = UUID_TO_BIN(?)
`, [keyframeId])
if(!kfRows.length) return({ ok: false, err: 'Keyframe not found' })
return({ ok: true, sim })
}
async validateSimulationOwner(userUuid, simulationUuid) {
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
const rows = await mysqlExecute(this.db, `
SELECT s.sim_id,
BIN_TO_UUID(s.sim_uuid) AS sim_uuid,
BIN_TO_UUID(s.sim_root_kf_uuid) AS sim_root_kf_uuid
FROM p42SIM.simulations s
INNER JOIN p42GUI.simowners o ON o.own_sim_uuid = s.sim_uuid
INNER JOIN p42GUI.users u ON o.own_usr_id = u.usr_id
WHERE u.usr_uuid = ?
AND s.sim_uuid = UUID_TO_BIN(?)
`, [userUuid, simulationUuid])
if(!rows.length) return({ ok: false, err: 'Simulation not found or access denied' })
return({ ok: true, sim: rows[0] })
}
async loadKeyframeAgents(keyframeId) {
const rows = await mysqlExecute(this.db, `
SELECT BIN_TO_UUID(ekfs_agent_id) AS agent_id, ekfs_gps_values
FROM p42SIM.edited_kf_store
WHERE ekfs_ekf_uuid = UUID_TO_BIN(?)
`, [keyframeId])
const agents = []
const errors = []
for(const row of rows) {
const parsed = parseGpsValues(row.ekfs_gps_values)
if(!parsed) {
errors.push(`Invalid GPS values for agent ${row.agent_id}`)
continue
}
agents.push({
id: row.agent_id,
position: parsed.position,
vector: parsed.vector,
})
}
if(errors.length) return({ ok: false, err: errors.join('; '), agents: [] })
return({ ok: true, agents })
}
}
-31
View File
@@ -1,31 +0,0 @@
#!/bin/sh
. /etc/p42/secrets.env
daemon=p42SimMaestro
logfile=maestro.log
pid=$(pgrep -f "$daemon")
if [ -z "$pid" ]
then
node "${daemon}.js" --debug > "$logfile" 2>&1 &
pid=$!
sleep 1
if kill -0 "$pid" 2>/dev/null
then
echo ""
echo "$daemon is now running with PID=$pid"
echo ""
else
echo ""
echo "Failed to start $daemon. Check maestro.log"
echo ""
fi
else
echo ""
echo "$daemon is already running with PID=$pid"
echo ""
fi
+1
View File
@@ -0,0 +1 @@
t
+43
View File
@@ -0,0 +1,43 @@
import { dispatchActions } from './dispatchActions.js'
import { dispatchEvents } from './dispatchEvents.js'
export function assembleHandlers(modules) {
const actions = {}
const tree = {}
const afterLogin = []
for(const mod of modules) {
if(mod.actions) Object.assign(actions, mod.actions)
if(typeof(mod.construct) === 'function') afterLogin.push(mod.construct)
if(!mod.eventHandlers) continue
for(const [channelPattern, byType] of Object.entries(mod.eventHandlers)) {
if(!tree[channelPattern]) tree[channelPattern] = {}
for(const [eventType, handler] of Object.entries(byType)) {
if(!tree[channelPattern][eventType]) tree[channelPattern][eventType] = []
const list = Array.isArray(handler) ? handler : [handler]
tree[channelPattern][eventType].push(...list)
}
}
}
return({
actionHandlers: actions,
eventHandlers: tree,
afterLogin,
})
}
export function createDispatchMessage({ eventHandlers, eventRules, actionRules }) {
return(async function dispatchMessage(redisCnx, msg, chan) {
if(msg.action && msg.eventType) {
console.warn(`[${redisCnx.redisId}] Message has both action and eventType on ${chan}`)
return(false)
}
if(msg.action) return(dispatchActions(redisCnx, msg, chan, actionRules(redisCnx)))
if(msg.eventType) {
const handlers = eventRules ? eventRules(redisCnx) : eventHandlers
return(dispatchEvents(redisCnx, msg, chan, handlers))
}
return(false)
})
}
+83
View File
@@ -0,0 +1,83 @@
import { replyToAction } from './publishActionReply.js'
function matchesActionsChannel(redisCnx, chan, channels) {
if(!Array.isArray(channels) || !channels.length) return(false)
for(const configured of channels) {
if(!configured) continue
if(redisCnx.fullChan(configured) === chan) return(true)
}
return(false)
}
export async function dispatchActions(redisCnx, msg, chan, rules) {
if(!matchesActionsChannel(redisCnx, chan, rules.channels)) return(false)
const action = msg.action
const sender = msg.sender ?? null
const cnxId = msg.cnxId ?? null
const reqid = ('reqid' in msg) ? msg.reqid.substr(0, 50) : null
const roles = Array.isArray(msg.roles) ? msg.roles : ['*']
if(!action || typeof(action) !== 'string') {
if(!sender) return(true)
replyToAction(redisCnx, {
action,
sender,
reqid,
cnxId,
success: false,
err: 'Missing or invalid action',
})
return(true)
}
if(!sender) {
console.warn(`[${redisCnx.redisId}] Action ${action} without sender on ${chan}`)
return(true)
}
if(redisCnx.accessRights && !redisCnx.accessRights.canDo(roles, action, sender)) {
replyToAction(redisCnx, {
action,
reqid,
sender,
cnxId,
success: false,
err: 'Unauthorized action !',
})
return(true)
}
const handler = redisCnx['action_'+action]
if(typeof(handler) !== 'function') {
replyToAction(redisCnx, {
action,
reqid,
sender,
cnxId,
success: false,
err: `Unknown action: ${action}`,
})
return(true)
}
if(redisCnx.debug) {
console.log(`[${redisCnx.redisId}] Dispatching action ${action} from ${sender}`)
}
try {
await handler.call(redisCnx, action, ('payload' in msg) ? msg.payload : null, reqid, sender, cnxId, roles)
} catch(err) {
console.error(`[${redisCnx.redisId}] Action ${action} failed:`, err)
replyToAction(redisCnx, {
action,
reqid,
sender,
cnxId,
success: false,
err: err.message ?? `${action} failed`,
})
}
return(true)
}
+35
View File
@@ -0,0 +1,35 @@
export function dispatchEvents(redisCnx, msg, chan, eventHandlers) {
const eventType = msg.eventType
const sender = msg.sender ?? null
const cnxId = msg.cnxId ?? null
if(!eventType || typeof(eventType) !== 'string') return(false)
let handled = false
for(const [channelPattern, byType] of Object.entries(eventHandlers ?? {})) {
if(!redisCnx.matchesChan(chan, channelPattern)) continue
const handlers = byType[eventType]
if(!handlers?.length) continue
for(const handle of handlers) {
try {
handle.call(redisCnx, msg, chan, sender, cnxId)
} catch(err) {
console.error(
`[${redisCnx.redisId}] Event ${eventType} on ${chan} failed:`,
err
)
}
}
handled = true
}
if(!handled && redisCnx.debug) {
console.log(`[${redisCnx.redisId}] Unhandled event ${eventType} on ${chan}`)
}
return(handled)
}
+64
View File
@@ -0,0 +1,64 @@
export function busReplyRoute(daemonBlock, meshName) {
if(!daemonBlock?.senderId) return(null)
const onArena = meshName === 'arena'
const actionsReply = onArena
? (daemonBlock.bus?.arena?.actionsReply ?? daemonBlock.ActionsReply)
: daemonBlock.ActionsReply
if(!actionsReply) return(null)
return({
senderId: daemonBlock.senderId,
actionsReply,
})
}
export function publishActionReply(redisCnx, options) {
const {
action,
reqid,
sender,
reply,
replyChannel,
} = options
reply.action = action
reply.sender = redisCnx.senderId
reply.cnxId = redisCnx.cnxId
if(reqid) reply.reqid = reqid
const chan = replyChannel.replace(/\[UID\]/g, sender)
.replace(/\[CUID\]/g, redisCnx.cnxId)
redisCnx.redisPublish(chan, reply)
}
export function replyToAction(redisCnx, options) {
const {
action,
reqid,
sender,
success,
payload,
err,
replyChannel,
} = options
const routeReplyChannel = replyChannel ?? redisCnx.actionsReply
if(!routeReplyChannel) {
console.error(`[${redisCnx.redisId}] Cannot resolve action reply route`)
return
}
const reply = { success }
if(err != null) reply.err = err
if(payload !== undefined) reply.payload = payload
publishActionReply(redisCnx, {
action,
reqid,
sender,
replyChannel: routeReplyChannel,
reply,
})
}
+120
View File
@@ -0,0 +1,120 @@
{
"accessRights": [
{
"canDo": [
"RELOADCONFIG",
"GETCONFIG"
],
"roles": [
"admin"
]
},
{
"canDo": [
"GETAGENTPOSITION",
"GETAGENTSINFRUSTUM",
"SUBSCRIBEFRUSTUM"
],
"roles": "*"
},
{
"canDo": [
"STARTSIMULATION",
"PAUSESIMULATION",
"STOPSIMULATION",
"GETSIMULATIONSSTATUS"
],
"roles": "*"
}
],
"gps": {
"primordialDaemon": true,
"ActionsChannel": "system:requests:gps",
"ActionsReply": "system:replies:[UID]",
"arenaActionsChannel": "arena:requests:[UID]",
"arenaActionsReply": "arena:replies:[UID]",
"GPSstorage": {
"agentHashKey": "system:gps:agent:[UID]",
"agentsIndexKey": "system:gps:agents",
"positionsStream": "system:gps:positions",
"streamMaxLen": 100000
},
"agentVectorChangeChannel": "arena:agents:*",
"collisionsChannel": "arena:agents:[UID]",
"lifecycle": {
"arenaChannel": "arena:lifecycle",
"godsReadyChannel": "arena:gods:ready"
},
"arenaStorage": {
"agentHashKey": "arena:agents:[UID]",
"agentsIndexKey": "arena:agents"
},
"senderId": "gps",
"nearMissDistance": 1,
"prismTimeHeight": 60,
"collisionTickMs": 100,
"prismRefreshLeadSeconds": 1
},
"maestro": {
"ActionsChannel": "system:requests:maestro",
"ActionsReply": "system:replies:[UID]",
"arenaActionsChannel": "arena:requests:[UID]",
"arenaActionsReply": "arena:replies:[UID]",
"senderId": "maestro",
"lifecycle": {
"arenaChannel": "arena:lifecycle",
"godsReadyChannel": "arena:gods:ready"
},
"systemLifecycleChannel": "system:maestro:lifecycle:[UID]",
"readyTimeoutMs": 30000
},
"observer": {
"primordialDaemon": false,
"ActionsChannel": "system:requests:observer",
"ActionsReply": "system:replies:[UID]",
"arenaActionsChannel": "arena:requests:[UID]",
"arenaActionsReply": "arena:replies:[UID]",
"FrustumEventsChannel": "system:observer:subscribed[CUID]:agents",
"senderId": "observer",
"scanIntervalMs": 300,
"lifecycle": {
"arenaChannel": "arena:lifecycle",
"godsReadyChannel": "arena:gods:ready"
}
},
"systemMesh": {
"redis": [
{
"redisId": "SYS_1",
"role": "primary",
"host": "127.0.0.1",
"tls": false,
"port": 6380,
"user": "",
"pass": "",
"chansNamespace": "system:",
"basePrefix": "messageBus:"
}
]
},
"arenaMesh": {
"redis": [
{
"redisId": "ARN_1",
"role": "primary",
"host": "127.0.0.1",
"tls": false,
"port": 6379,
"user": "",
"pass": "",
"chansNamespace": "arena:",
"basePrefix": "messageBus:"
}
]
},
"mysql": {
"socketPath": "/var/run/mysqld/mysqld.sock",
"guiDatabase": "p42GUI",
"simDatabase": "p42SIM"
}
}
+33 -19
View File
@@ -60,9 +60,12 @@
"gps": {
"type": "object",
"properties": {
"gpsActionsChannel": { "type": "string" },
"gpsActionsReply": { "type": "string" },
"storage": {
"primordialDaemon": { "type": "boolean" },
"ActionsChannel": { "type": "string" },
"ActionsReply": { "type": "string" },
"arenaActionsChannel": { "type": "string" },
"arenaActionsReply": { "type": "string" },
"GPSstorage": {
"type": "object",
"properties": {
"agentHashKey": { "type": "string" },
@@ -107,9 +110,11 @@
"prismRefreshLeadSeconds": { "type": "number", "minimum": 0 }
},
"required": [
"gpsActionsChannel",
"gpsActionsReply",
"storage",
"ActionsChannel",
"ActionsReply",
"arenaActionsChannel",
"arenaActionsReply",
"GPSstorage",
"agentVectorChangeChannel",
"collisionsChannel"
]
@@ -117,8 +122,10 @@
"maestro": {
"type": "object",
"properties": {
"maestroActionsChannel": { "type": "string" },
"maestroActionsReply": { "type": "string" },
"ActionsChannel": { "type": "string" },
"ActionsReply": { "type": "string" },
"arenaActionsChannel": { "type": "string" },
"arenaActionsReply": { "type": "string" },
"senderId": { "type": "string" },
"lifecycle": {
"type": "object",
@@ -131,15 +138,13 @@
"godsReadyChannel"
]
},
"expectedGods": {
"type": "array",
"items": { "type": "string" }
},
"readyTimeoutMs": { "type": "integer", "minimum": 1000 }
},
"required": [
"maestroActionsChannel",
"maestroActionsReply"
"ActionsChannel",
"ActionsReply",
"arenaActionsChannel",
"arenaActionsReply"
]
},
"mysql": {
@@ -148,7 +153,8 @@
"socketPath": { "type": "string" },
"host": { "type": "string" },
"port": { "type": "integer" },
"database": { "type": "string" },
"guiDatabase": { "type": "string" },
"simDatabase": { "type": "string" },
"connectionLimit": { "type": "integer", "minimum": 1 }
},
"required": []
@@ -156,9 +162,14 @@
"observer": {
"type": "object",
"properties": {
"observerActionsChannel": { "type": "string" },
"observerActionsReply": { "type": "string" },
"primordialDaemon": { "type": "boolean" },
"ActionsChannel": { "type": "string" },
"ActionsReply": { "type": "string" },
"arenaActionsChannel": { "type": "string" },
"arenaActionsReply": { "type": "string" },
"FrustumEventsChannel": { "type": "string" },
"senderId": { "type": "string" },
"scanIntervalMs": { "type": "integer", "minimum": 50 },
"lifecycle": {
"type": "object",
"properties": {
@@ -172,8 +183,11 @@
}
},
"required": [
"observerActionsChannel",
"observerActionsReply"
"ActionsChannel",
"ActionsReply",
"FrustumEventsChannel",
"arenaActionsChannel",
"arenaActionsReply"
]
},
"systemMesh": {
+120
View File
@@ -0,0 +1,120 @@
{
"accessRights": [
{
"canDo": [
"RELOADCONFIG",
"GETCONFIG"
],
"roles": [
"admin"
]
},
{
"canDo": [
"GETAGENTPOSITION",
"GETAGENTSINFRUSTUM",
"SUBSCRIBEFRUSTUM"
],
"roles": "*"
},
{
"canDo": [
"STARTSIMULATION",
"PAUSESIMULATION",
"STOPSIMULATION",
"GETSIMULATIONSSTATUS"
],
"roles": "*"
}
],
"gps": {
"primordialDaemon": true,
"ActionsChannel": "system:requests:gps",
"ActionsReply": "system:replies:[UID]",
"arenaActionsChannel": "arena:requests:[UID]",
"arenaActionsReply": "arena:replies:[UID]",
"GPSstorage": {
"agentHashKey": "system:gps:agent:[UID]",
"agentsIndexKey": "system:gps:agents",
"positionsStream": "system:gps:positions",
"streamMaxLen": 100000
},
"agentVectorChangeChannel": "arena:agents:*",
"collisionsChannel": "arena:agents:[UID]",
"lifecycle": {
"arenaChannel": "arena:lifecycle",
"godsReadyChannel": "arena:gods:ready"
},
"arenaStorage": {
"agentHashKey": "arena:agents:[UID]",
"agentsIndexKey": "arena:agents"
},
"senderId": "gps",
"nearMissDistance": 1,
"prismTimeHeight": 60,
"collisionTickMs": 100,
"prismRefreshLeadSeconds": 1
},
"maestro": {
"ActionsChannel": "system:requests:maestro",
"ActionsReply": "system:replies:[UID]",
"arenaActionsChannel": "arena:requests:[UID]",
"arenaActionsReply": "arena:replies:[UID]",
"senderId": "maestro",
"lifecycle": {
"arenaChannel": "arena:lifecycle",
"godsReadyChannel": "arena:gods:ready"
},
"systemLifecycleChannel": "system:maestro:lifecycle:[UID]",
"readyTimeoutMs": 30000
},
"observer": {
"primordialDaemon": false,
"ActionsChannel": "system:requests:observer",
"ActionsReply": "system:replies:[UID]",
"FrustumEventsChannel": "system:observer:subscribed[CUID]:agents",
"arenaActionsChannel": "arena:requests:[UID]",
"arenaActionsReply": "arena:replies:[UID]",
"senderId": "observer",
"scanIntervalMs": 300,
"lifecycle": {
"arenaChannel": "arena:lifecycle",
"godsReadyChannel": "arena:gods:ready"
}
},
"systemMesh": {
"redis": [
{
"redisId": "SYS_1",
"role": "primary",
"host": "127.0.0.1",
"tls": false,
"port": 6380,
"user": "",
"pass": "",
"chansNamespace": "system:",
"basePrefix": "messageBus:"
}
]
},
"arenaMesh": {
"redis": [
{
"redisId": "ARN_1",
"role": "primary",
"host": "127.0.0.1",
"tls": false,
"port": 6379,
"user": "",
"pass": "",
"chansNamespace": "arena:",
"basePrefix": "messageBus:"
}
]
},
"mysql": {
"socketPath": "/var/run/mysqld/mysqld.sock",
"guiDatabase": "test_p42GUI",
"simDatabase": "test_p42SIM"
}
}
+22
View File
@@ -0,0 +1,22 @@
#!/bin/sh
# resolveConfigPath BASE PATH
# Turn PATH into an absolute path; relative segments are resolved from BASE.
resolveConfigPath() {
_base=$1
_path=$2
case "$_path" in
/*)
printf '%s\n' "$_path"
;;
*)
_dir=$(dirname "$_path")
_name=$(basename "$_path")
if [ "$_dir" = . ]; then
printf '%s\n' "$_base/$_name"
else
printf '%s\n' "$(cd "$_base/$_dir" && pwd)/$_name"
fi
;;
esac
}
-29
View File
@@ -1,29 +0,0 @@
import mysql from 'mysql2/promise'
export function resolveMysqlCredentials(config = {}) {
const user = process.env.user
const password = process.env.mysql_pass
if(!user || !password) {
throw new Error('Missing MySQL credentials: set user and mysql_pass in environment')
}
return({
socketPath: config.socketPath,
host: config.host,
port: config.port,
user,
password,
database: config.database ?? 'p42GUI',
waitForConnections: true,
connectionLimit: config.connectionLimit ?? 5,
queueLimit: 0,
})
}
export async function createMysqlPool(config) {
return(await mysql.createPool(resolveMysqlCredentials(config)))
}
export async function mysqlExecute(pool, query, values = []) {
const [rows] = await pool.execute(query, values)
return(rows)
}
+2 -1
View File
@@ -2,9 +2,10 @@
"name": "p42GodDaemons",
"type": "module",
"dependencies": {
"@p42/p42modules": "^0.1.0",
"ajv": "^8.12.0",
"mysql2": "^3.11.0",
"redis": "^4.3.0",
"uuid": "^14.0.0",
"yargs": "^17.7.2"
}
}
+13 -4
View File
@@ -1,4 +1,5 @@
import redis from 'redis'
import os from 'node:os'
export class RedisConnexion {
@@ -9,9 +10,13 @@ export class RedisConnexion {
this.redisConfig = this.config
this.meshName = options.meshName
this.meshModule = options.meshModule ?? null
this.senderId = options.senderId ?? null
this.actionsReply = options.actionsReply ?? null
if(this.meshModule?.meshActions) Object.assign(this, this.meshModule.meshActions)
this.afterLoginMethods = this.meshModule?.afterLoginMethods ?? []
this.cnxId = os.hostname() + ':' + process.pid + ':' + Date.now() + ':' + Math.random().toString(36).substring(2, 15)
if(this.meshModule?.actionHandlers) Object.assign(this, this.meshModule.actionHandlers)
this.afterLogin = this.meshModule?.afterLogin ?? []
this.redisClient = redis.createClient({
socket: {
@@ -56,7 +61,7 @@ export class RedisConnexion {
console.log(`[${this.redisConfig.redisId}] Redis ${this.redisConfig.redisId} time:`, redisTime)
}
for(const method of this.afterLoginMethods){
for(const method of this.afterLogin){
if(typeof method != 'function') continue
method(this)
}
@@ -282,7 +287,11 @@ export class RedisConnexion {
}
if(typeof(this.meshModule?.dispatchMessage) === 'function') {
this.meshModule.dispatchMessage(this, msg, chan)
try {
await this.meshModule.dispatchMessage(this, msg, chan)
} catch(err) {
console.error(`[${this.redisConfig.redisId}] dispatchMessage failed on ${chan}:`, err)
}
}
}
+75
View File
@@ -0,0 +1,75 @@
#!/bin/bash
GOD_FOLDERS="Maestro GPS Observer"
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=lib/resolveConfigPath.sh
. "$ROOT/lib/resolveConfigPath.sh"
CONFIG="config.json"
usage() {
echo "Usage: $(basename "$0") [-c|--config PATH]" >&2
echo " PATH is relative to $ROOT unless absolute." >&2
}
while [ $# -gt 0 ]; do
case "$1" in
-c|--config)
if [ -z "${2:-}" ]; then
echo "Missing value for $1" >&2
usage
exit 1
fi
CONFIG="$2"
shift 2
;;
-h|--help)
usage
exit 0
;;
*)
CONFIG="$1"
shift
;;
esac
done
CONFIG="$(resolveConfigPath "$ROOT" "$CONFIG")"
if [ ! -f "$CONFIG" ]; then
echo "Config file not found: $CONFIG" >&2
exit 1
fi
failed=0
for folder in $GOD_FOLDERS; do
dir="$ROOT/$folder"
if [ ! -d "$dir" ]; then
echo "Missing daemon folder: $dir" >&2
failed=1
continue
fi
shopt -s nullglob
scripts=("$dir"/start*.sh)
shopt -u nullglob
if [ ${#scripts[@]} -eq 0 ]; then
echo "No start script in $dir" >&2
failed=1
continue
fi
if [ ${#scripts[@]} -gt 1 ]; then
echo "Multiple start scripts in $dir, using ${scripts[0]}" >&2
fi
echo "=== Starting $folder (config=$CONFIG) ==="
(cd "$dir" && bash "${scripts[0]}" "$CONFIG") || {
echo "Failed to start $folder" >&2
failed=1
}
done
exit $failed
Executable
+38
View File
@@ -0,0 +1,38 @@
#!/bin/bash
GOD_FOLDERS="Maestro GPS Observer"
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
failed=0
read -ra folders <<< "$GOD_FOLDERS"
for((i=${#folders[@]}-1; i>=0; i--)); do
folder="${folders[i]}"
dir="$ROOT/$folder"
if [ ! -d "$dir" ]; then
echo "Missing daemon folder: $dir" >&2
failed=1
continue
fi
shopt -s nullglob
scripts=("$dir"/stop*.sh)
shopt -u nullglob
if [ ${#scripts[@]} -eq 0 ]; then
echo "No stop script in $dir" >&2
failed=1
continue
fi
if [ ${#scripts[@]} -gt 1 ]; then
echo "Multiple stop scripts in $dir, using ${scripts[0]}" >&2
fi
echo "=== Stopping $folder ==="
(cd "$dir" && bash "${scripts[0]}") || {
echo "Failed to stop $folder" >&2
failed=1
}
done
exit $failed
+11
View File
@@ -0,0 +1,11 @@
# Test DBs: point all gods at the same config (or edit config.json mysql section).
# Example:
# ./startAllGods.sh --config /opt/p42GodDaemons/config.json
# ./Maestro/startMaestro.sh /opt/p42GodDaemons/config.json
clear; node test.js maestro1 \
--config ../config.json \
--guiDatabase test_p42GUI --simDatabase test_p42SIM \
--fakeAgentsReady --fakeGpsReady \
--userUuid a4f33373-6adf-4d2d-9a6d-7fa0abf8b01f \
--simulationUuid 0x019ec742e12175c685a97bf9300b6b49
+289
View File
@@ -0,0 +1,289 @@
import { MySQLClient } from '@p42/p42modules'
import { SimRepository } from '../../Maestro/simRepository.js'
function agentHashKey(template, agentId) {
return(template.replace(/\[UID\]/g, agentId))
}
function hashFieldValue(raw) {
if(raw == null) return(null)
if(typeof(raw) !== 'string') return(raw)
try { return(JSON.parse(raw)) }
catch { return(raw) }
}
function valuesEqual(a, b) {
if(a === b) return(true)
if(a == null || b == null) return(false)
if(typeof(a) !== 'object' && typeof(b) !== 'object') return(a == b)
if(typeof(a) !== typeof(b)) return(false)
if(typeof(a) !== 'object') return(false)
if(Array.isArray(a) !== Array.isArray(b)) return(false)
const keysA = Object.keys(a)
const keysB = Object.keys(b)
if(keysA.length !== keysB.length) return(false)
for(const key of keysA) {
if(!valuesEqual(a[key], b[key])) return(false)
}
return(true)
}
const RESERVED_HASH_FIELDS = new Set(['position', 'vector', 'speed', 'segment'])
function vectorsEqual(a, b) {
for(const axis of ['x', 'y', 'z']) {
if(a[axis] !== b[axis]) return(false)
}
return(true)
}
async function findSimulationFixture(ctx) {
const { guiDatabase, simDatabase } = ctx.databases
const rows = await MySQLClient.poolExecute(ctx.db, `
SELECT usr_uuid AS user_uuid,
BIN_TO_UUID(sim_uuid) AS simulation_uuid
FROM \`${guiDatabase}\`.users
INNER JOIN \`${guiDatabase}\`.simowners ON own_usr_id = usr_id
INNER JOIN \`${simDatabase}\`.simulations ON own_sim_uuid = sim_uuid
INNER JOIN \`${simDatabase}\`.edited_kf_store ON ekfs_ekf_uuid = sim_root_kf_uuid
GROUP BY usr_uuid, sim_uuid
HAVING COUNT(ekfs_agent_id) > 0
LIMIT 1
`)
if(!rows.length) {
throw(new Error(
'No simulation fixture found in MySQL (need user-owned sim with root keyframe agents). '
+ 'Pass --userUuid, --simulationUuid explicitly, or use --guiDatabase / --simDatabase for a test DB.'
))
}
return({
userUuid: rows[0].user_uuid,
simulationUuid: rows[0].simulation_uuid,
})
}
function godsReadyChannel(config) {
return(config.maestro?.lifecycle?.godsReadyChannel
?? config.gps?.lifecycle?.godsReadyChannel
?? 'arena:gods:ready')
}
async function publishFakeReadyToStart(ctx, { sender, simulationId, agentIds }) {
const { arenaCnx } = ctx
const payload = {
success: true,
simulationId,
err: null,
}
if(agentIds) payload.agentIds = agentIds
await arenaCnx.redisPublish(godsReadyChannel(ctx.config), {
eventType: 'readyToStart',
sender,
payload,
})
}
async function publishFakeReadies(ctx, simulationId, agentIds) {
const { argv, config, log } = ctx
if(argv.fakeGpsReady) {
const senderId = config.gps?.senderId ?? 'gps'
log('action', `Faking GPS readyToStart (sender=${senderId})...`)
await publishFakeReadyToStart(ctx, {
sender: senderId,
simulationId,
agentIds,
})
log('success', 'Published fake GPS readyToStart')
}
if(argv.fakeAgentsReady) {
log('action', `Faking readyToStart for ${agentIds.length} agent(s)...`)
for(const agentId of agentIds) {
await publishFakeReadyToStart(ctx, {
sender: agentId,
simulationId,
})
}
log('success', `Published fake readyToStart for ${agentIds.length} agent(s)`)
}
}
function waitForLifecycleEvent(ctx, eventType, timeoutMs) {
return(new Promise((resolve, reject) => {
const lifecyclePattern = ctx.config.maestro.lifecycle.arenaChannel
const timer = setTimeout(() => {
ctx.offArenaMessage(handler)
reject(new Error(`Timeout waiting for ${eventType} on ${lifecyclePattern} (${timeoutMs}ms)`))
}, timeoutMs)
const handler = (msg, chan) => {
if(!ctx.arenaCnx.matchesChan(chan, lifecyclePattern)) return
if(msg.eventType !== eventType) return
clearTimeout(timer)
ctx.offArenaMessage(handler)
resolve(msg)
}
ctx.onArenaMessage(handler)
}))
}
export function configureYargs(yargsBuilder) {
return(yargsBuilder.options({
userUuid: {
describe: 'User UUID (dashed or 0x-prefixed hex; auto-discovered if omitted)',
type: 'string',
},
simulationUuid: {
describe: 'Simulation UUID (dashed or 0x-prefixed hex; auto-discovered if omitted)',
type: 'string',
},
timeout: {
describe: 'Milliseconds to wait for onYourMarks',
default: 15000,
type: 'number',
},
fakeAgentsReady: {
describe: 'After onYourMarks, publish fake readyToStart for each seeded agent',
type: 'boolean',
default: false,
},
fakeGpsReady: {
describe: 'After onYourMarks, publish fake GPS readyToStart',
type: 'boolean',
default: false,
},
}))
}
export async function run(ctx) {
const { log, argv, config, systemCnx, arenaCnx, normalizeUuid } = ctx
const arenaStorage = config.gps?.arenaStorage ?? {
agentHashKey: 'arena:agents:[UID]',
agentsIndexKey: 'arena:agents',
}
log('action', 'Resolving simulation fixture from MySQL...')
let userUuid = argv.userUuid ? normalizeUuid(argv.userUuid) : undefined
let simulationUuid = argv.simulationUuid ? normalizeUuid(argv.simulationUuid) : undefined
if(!userUuid || !simulationUuid) {
const fixture = await findSimulationFixture(ctx)
userUuid = userUuid ?? fixture.userUuid
simulationUuid = simulationUuid ?? fixture.simulationUuid
}
log('action', `User: ${userUuid}`)
log('action', `Simulation: ${simulationUuid}`)
const simRepo = new SimRepository(ctx.db, ctx.databases, false)
const access = await simRepo.validateSimulationAccess(userUuid, simulationUuid)
if(!access.ok) throw(new Error(`Simulation access check failed: ${access.err}`))
const keyframeId = access.sim.sim_root_kf_uuid
log('action', `Root keyframe: ${keyframeId}`)
const agentsResult = await simRepo.loadKeyframeAgents(keyframeId)
if(!agentsResult.ok) throw(new Error(`Failed to load keyframe agents: ${agentsResult.err}`))
if(!agentsResult.agents.length) throw(new Error('Keyframe has no agents with valid GPS values'))
log('success', `Loaded ${agentsResult.agents.length} agent(s) from MySQL`)
const expectedById = new Map(agentsResult.agents.map(a => [a.id, a]))
const lifecycleWait = waitForLifecycleEvent(ctx, 'onYourMarks', argv.timeout)
const reqid = `maestro1-${Date.now()}`
const actionsChan = config.maestro.ActionsChannel
log('action', `Publishing STARTSIMULATION on ${actionsChan} (reqid=${reqid})...`)
await systemCnx.redisPublish(actionsChan, {
action: 'STARTSIMULATION',
reqid,
sender: userUuid,
roles: ['*'],
payload: {
simulationUuid,
infraId: null,
},
})
log('action', `Waiting for onYourMarks on ${config.maestro.lifecycle.arenaChannel}...`)
const lifecycleMsg = await lifecycleWait
const payload = lifecycleMsg.payload ?? {}
if(payload.simulationId !== simulationUuid) {
throw(new Error(`onYourMarks simulationId mismatch: expected ${simulationUuid}, got ${payload.simulationId}`))
}
log('success', `Received onYourMarks for simulationId=${payload.simulationId}`)
const expectedIds = [...expectedById.keys()]
if(argv.fakeAgentsReady || argv.fakeGpsReady) {
await publishFakeReadies(ctx, payload.simulationId, expectedIds)
}
log('action', 'Reading arena store...')
const indexIds = await arenaCnx.redisSmembers(arenaStorage.agentsIndexKey)
const sortedExpectedIds = [...expectedIds].sort()
const actualIds = [...indexIds].sort()
if(actualIds.length !== sortedExpectedIds.length) {
throw(new Error(`Agent index count mismatch: expected ${sortedExpectedIds.length}, got ${actualIds.length}`))
}
for(let i = 0; i < sortedExpectedIds.length; i++) {
if(actualIds[i] !== sortedExpectedIds[i]) {
throw(new Error(`Agent index mismatch at ${i}: expected ${sortedExpectedIds[i]}, got ${actualIds[i]}`))
}
}
log('success', `Arena agents index contains ${actualIds.length} agent(s)`)
for(const agentId of sortedExpectedIds) {
const key = agentHashKey(arenaStorage.agentHashKey, agentId)
const hash = await arenaCnx.redisHgetall(key)
const expected = expectedById.get(agentId)
const position = hashFieldValue(hash.position)
const vector = hashFieldValue(hash.vector)
if(!position || !vector) {
throw(new Error(`Agent ${agentId}: missing position or vector in arena hash ${key}`))
}
if(!vectorsEqual(position, expected.position)) {
throw(new Error(`Agent ${agentId}: position mismatch (MySQL vs arena store)`))
}
if(!vectorsEqual(vector, expected.vector)) {
throw(new Error(`Agent ${agentId}: vector mismatch (MySQL vs arena store)`))
}
const storeExpected = expected.store ?? {}
for(const [field, expVal] of Object.entries(storeExpected)) {
const actualVal = hashFieldValue(hash[field])
if(!valuesEqual(actualVal, expVal)) {
throw(new Error(`Agent ${agentId}: store field "${field}" mismatch (MySQL vs arena store)`))
}
}
for(const field of Object.keys(hash)) {
if(RESERVED_HASH_FIELDS.has(field)) continue
if(!(field in storeExpected)) {
throw(new Error(`Agent ${agentId}: unexpected store field "${field}" in arena hash ${key}`))
}
}
log('success', `Agent ${agentId}: position, vector, and store values match MySQL`)
}
if(argv.fakeAgentsReady || argv.fakeGpsReady) {
log('success', 'Arena store seeded correctly from MySQL; fake prepare acks published')
} else {
log('success', 'Arena store seeded correctly from MySQL (Maestro will wait for agent + primordial daemon readyToStart until timeout)')
}
}
+212
View File
@@ -0,0 +1,212 @@
import yargs from 'yargs/yargs'
import { hideBin } from 'yargs/helpers'
import { pathToFileURL } from 'url'
import { fileURLToPath } from 'url'
import { dirname, join, resolve } from 'path'
import { RedisConnexion } from '../redisConnexion.js'
import { configHelper } from '../configHelper.js'
import { MySQLClient } from '@p42/p42modules'
const __dirname = dirname(fileURLToPath(import.meta.url))
const __filename = fileURLToPath(import.meta.url)
const UUID_DASHED_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
const UUID_HEX32_RE = /^[0-9a-f]{32}$/i
export function normalizeUuid(value) {
if(typeof(value) !== 'string') {
throw(new Error('UUID must be a string'))
}
const trimmed = value.trim()
if(!trimmed) throw(new Error('UUID must be a non-empty string'))
if(UUID_DASHED_RE.test(trimmed)) {
return(trimmed.toLowerCase())
}
let hex = trimmed
if(/^0x/i.test(hex)) hex = hex.slice(2)
if(!UUID_HEX32_RE.test(hex)) {
throw(new Error(
`Invalid UUID: ${value} (expected dashed form or 32-char hex, optionally prefixed with 0x)`
))
}
hex = hex.toLowerCase()
return(`${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`)
}
const LOG_COLORS = {
action: '\x1b[37m',
success: '\x1b[32m',
error: '\x1b[31m',
reset: '\x1b[0m',
}
function createTestMesh(handlers) {
return({
dispatchMessage(_cnx, msg, chan) {
for(const handler of handlers) handler(msg, chan)
},
})
}
function buildBaseParser() {
return(yargs(hideBin(process.argv))
.usage('$0 <module> [options]')
.command('$0 <module>', 'Run a GodDaemons test module', y => y
.positional('module', {
describe: 'Test module name (e.g. maestro1)',
type: 'string',
})
)
.options({
config: {
alias: 'c',
describe: 'Path to config.json',
default: join(__dirname, '..', 'config.json'),
type: 'string',
},
guiDatabase: {
describe: 'Override mysql.guiDatabase for this run',
type: 'string',
},
simDatabase: {
describe: 'Override mysql.simDatabase for this run',
type: 'string',
},
})
.help()
.version(false)
)
}
function meshRedisConn(mesh, meshName, rootConfig, meshModule) {
const { redis, ...meshConfig } = mesh
return(redis.map(cfg =>
new RedisConnexion({
debug: false,
config: { ...cfg, ...meshConfig, ...rootConfig },
redisId: cfg.redisId,
meshName,
meshModule,
})
))
}
async function loadTestModule(moduleName) {
const modulePath = join(__dirname, 'modules', `${moduleName}.js`)
try {
return(await import(pathToFileURL(modulePath).href))
} catch(err) {
if(err.code === 'ERR_MODULE_NOT_FOUND') {
throw(new Error(`Test module not found: ${moduleName} (expected ${modulePath})`))
}
throw(err)
}
}
async function main() {
const preArgv = buildBaseParser().parseSync()
const moduleName = preArgv.module ?? preArgv._[0]
if(!moduleName) {
buildBaseParser().showHelp()
process.exit(1)
}
const testModule = await loadTestModule(moduleName)
let parser = buildBaseParser()
if(typeof(testModule.configureYargs) === 'function') {
parser = testModule.configureYargs(parser)
}
const argv = parser.parseSync()
const resolvedModule = argv.module ?? argv._[0]
const cfgh = new configHelper({ localfile: argv.config })
await cfgh.fetchConfigFile()
const config = cfgh.config
if(argv.guiDatabase) config.mysql.guiDatabase = argv.guiDatabase
if(argv.simDatabase) config.mysql.simDatabase = argv.simDatabase
const arenaHandlers = new Set()
const testMesh = createTestMesh(arenaHandlers)
const systemConns = meshRedisConn(config.systemMesh, 'system', config, null)
const arenaConns = meshRedisConn(config.arenaMesh, 'arena', config, testMesh)
const systemCnx = systemConns.find(c => c.redisConfig.role === 'primary') ?? systemConns[0]
const arenaCnx = arenaConns.find(c => c.redisConfig.role === 'primary') ?? arenaConns[0]
const log = (type, message) => {
const color = LOG_COLORS[type] ?? LOG_COLORS.action
console.log(`${color}${message}${LOG_COLORS.reset}`)
}
log('action', `Loading test module: ${resolvedModule}`)
log('action', `Config: ${argv.config}`)
const databases = MySQLClient.resolveDatabases(config.mysql)
if(argv.guiDatabase || argv.simDatabase) {
log('action', `MySQL databases: gui=${databases.guiDatabase}, sim=${databases.simDatabase}`)
}
log('action', 'Connecting to Redis (system)...')
await systemCnx.redisLogin()
log('success', `System Redis connected (${systemCnx.redisConfig.host}:${systemCnx.redisConfig.port})`)
log('action', 'Connecting to Redis (arena)...')
await arenaCnx.redisLogin()
await arenaCnx.redisChansStart()
log('success', `Arena Redis connected (${arenaCnx.redisConfig.host}:${arenaCnx.redisConfig.port})`)
log('action', 'Connecting to MySQL...')
const db = await MySQLClient.createPool(config.mysql)
log('success', 'MySQL pool ready')
const ctx = {
argv,
config,
databases,
db,
systemCnx,
arenaCnx,
log,
normalizeUuid,
onArenaMessage(handler) {
arenaHandlers.add(handler)
},
offArenaMessage(handler) {
arenaHandlers.delete(handler)
},
}
let exitCode = 0
try {
if(typeof(testModule.run) !== 'function') {
throw(new Error(`Test module ${resolvedModule} must export a run(ctx) function`))
}
await testModule.run(ctx)
log('success', `Test module ${resolvedModule} finished`)
} catch(err) {
log('error', err.message ?? String(err))
exitCode = 1
} finally {
await db.end()
await systemCnx.redisClient.quit()
await arenaCnx.redisClient.quit()
if(arenaCnx.redisSubscriber) await arenaCnx.redisSubscriber.quit()
}
process.exit(exitCode)
}
const isMain = process.argv[1] && resolve(__filename) === resolve(process.argv[1])
if(isMain) {
main().catch(err => {
console.error(`${LOG_COLORS.error}${err.message ?? err}${LOG_COLORS.reset}`)
process.exit(1)
})
}
+2
View File
@@ -0,0 +1,2 @@
import { v7 as uuidv7 } from 'uuid'
console.log(uuidv7().replaceAll('-', ''))