make kestrel a tak server, so that it can send and receive pois as cots data

server/utils/liveSessions.js

@@ -1,47 +1,79 @@
 import { closeRouter, getProducer, getTransport } from './mediasoup.js'
+import { acquire } from './asyncLock.js'
+import { LIVE_SESSION_TTL_MS } from './constants.js'
 
-const TTL_MS = 60_000
 const sessions = new Map()
 
-export const createSession = (userId, label = '') => {
-  const id = crypto.randomUUID()
-  const session = {
-    id,
-    userId,
-    label: (label || 'Live').trim() || 'Live',
-    lat: 0,
-    lng: 0,
-    updatedAt: Date.now(),
-    routerId: null,
-    producerId: null,
-    transportId: null,
-  }
-  sessions.set(id, session)
-  return session
+export const createSession = async (userId, label = '') => {
+  return acquire(`session-create-${userId}`, async () => {
+    const id = crypto.randomUUID()
+    const session = {
+      id,
+      userId,
+      label: (label || 'Live').trim() || 'Live',
+      lat: 0,
+      lng: 0,
+      updatedAt: Date.now(),
+      routerId: null,
+      producerId: null,
+      transportId: null,
+    }
+    sessions.set(id, session)
+    return session
+  })
+}
+
+/**
+ * Atomically get existing active session or create new one for user.
+ * @param {string} userId - User ID
+ * @param {string} label - Session label
+ * @returns {Promise<object>} Session object
+ */
+export const getOrCreateSession = async (userId, label = '') => {
+  return acquire(`session-get-or-create-${userId}`, async () => {
+    const now = Date.now()
+    for (const s of sessions.values()) {
+      if (s.userId === userId && now - s.updatedAt <= LIVE_SESSION_TTL_MS) {
+        return s
+      }
+    }
+    return await createSession(userId, label)
+  })
 }
 
 export const getLiveSession = id => sessions.get(id)
 
-export const getActiveSessionByUserId = (userId) => {
-  const now = Date.now()
-  for (const s of sessions.values()) {
-    if (s.userId === userId && now - s.updatedAt <= TTL_MS) return s
-  }
+export const getActiveSessionByUserId = async (userId) => {
+  return acquire(`session-get-${userId}`, async () => {
+    const now = Date.now()
+    for (const s of sessions.values()) {
+      if (s.userId === userId && now - s.updatedAt <= LIVE_SESSION_TTL_MS) return s
+    }
+  })
 }
 
-export const updateLiveSession = (id, updates) => {
-  const session = sessions.get(id)
-  if (!session) return
-  const now = Date.now()
-  if (Number.isFinite(updates.lat)) session.lat = updates.lat
-  if (Number.isFinite(updates.lng)) session.lng = updates.lng
-  if (updates.routerId !== undefined) session.routerId = updates.routerId
-  if (updates.producerId !== undefined) session.producerId = updates.producerId
-  if (updates.transportId !== undefined) session.transportId = updates.transportId
-  session.updatedAt = now
+export const updateLiveSession = async (id, updates) => {
+  return acquire(`session-update-${id}`, async () => {
+    const session = sessions.get(id)
+    if (!session) {
+      throw new Error('Session not found')
+    }
+    const now = Date.now()
+    if (Number.isFinite(updates.lat)) session.lat = updates.lat
+    if (Number.isFinite(updates.lng)) session.lng = updates.lng
+    if (updates.routerId !== undefined) session.routerId = updates.routerId
+    if (updates.producerId !== undefined) session.producerId = updates.producerId
+    if (updates.transportId !== undefined) session.transportId = updates.transportId
+    session.updatedAt = now
+    return session
+  })
 }
 
-export const deleteLiveSession = id => sessions.delete(id)
+export const deleteLiveSession = async (id) => {
+  await acquire(`session-delete-${id}`, async () => {
+    sessions.delete(id)
+  })
+}
 
 export const clearSessions = () => sessions.clear()
 
@@ -62,31 +94,33 @@ const cleanupSession = async (session) => {
 }
 
 export const getActiveSessions = async () => {
-  const now = Date.now()
-  const active = []
-  const expired = []
+  return acquire('get-active-sessions', async () => {
+    const now = Date.now()
+    const active = []
+    const expired = []
 
-  for (const session of sessions.values()) {
-    if (now - session.updatedAt <= TTL_MS) {
-      active.push({
-        id: session.id,
-        userId: session.userId,
-        label: session.label,
-        lat: session.lat,
-        lng: session.lng,
-        updatedAt: session.updatedAt,
-        hasStream: Boolean(session.producerId),
-      })
+    for (const session of sessions.values()) {
+      if (now - session.updatedAt <= LIVE_SESSION_TTL_MS) {
+        active.push({
+          id: session.id,
+          userId: session.userId,
+          label: session.label,
+          lat: session.lat,
+          lng: session.lng,
+          updatedAt: session.updatedAt,
+          hasStream: Boolean(session.producerId),
+        })
+      }
+      else {
+        expired.push(session)
+      }
     }
-    else {
-      expired.push(session)
+
+    for (const session of expired) {
+      await cleanupSession(session)
+      sessions.delete(session.id)
     }
-  }
 
-  for (const session of expired) {
-    await cleanupSession(session)
-    sessions.delete(session.id)
-  }
-
-  return active
+    return active
+  })
 }