CRM_ENVR/signaling-server.mjs

import { createServer } from "http"
import { Server } from "socket.io"
import { SignJWT, jwtVerify } from "jose"
import pg from "pg"

const PORT = process.env.SIGNALING_PORT || 3007
const rawSecret = process.env.JWT_SECRET
if (!rawSecret) throw new Error("JWT_SECRET environment variable is required")
const JWT_SECRET = new TextEncoder().encode(rawSecret)
const DATABASE_URL = process.env.DATABASE_URL
if (!DATABASE_URL) throw new Error("DATABASE_URL environment variable is required")

const pool = new pg.Pool({ connectionString: DATABASE_URL })

const httpServer = createServer()
const io = new Server(httpServer, { cors: { origin: "*", methods: ["GET", "POST"] } })

const onlineUsers = new Map()
const rooms = new Map()

io.use((socket, next) => {
  const token = socket.handshake.auth?.token
  if (token) {
    jwtVerify(token, JWT_SECRET)
      .then(({ payload }) => {
        socket.data.userId = payload.userId
        socket.data.role = payload.role
        next()
      })
      .catch(() => {
        socket.data.userId = null
        socket.data.role = null
        next()
      })
  } else {
    socket.data.userId = null
    socket.data.role = null
    next()
  }
})

io.on("connection", (socket) => {
  const userId = socket.data.userId

  if (userId) {
    onlineUsers.set(userId, socket.id)
    socket.broadcast.emit("user:online", { userId })
  }

  if (userId) {
    socket.on("user:lookup", async ({ phone }, callback) => {
      try {
        const result = await pool.query(
          `SELECT id, username, first_name, last_name, phone, avatar_url
           FROM users
           WHERE phone = $1 AND deleted_at IS NULL
           LIMIT 1`,
          [phone],
        )
        if (result.rows.length > 0) {
          const user = result.rows[0]
          callback({
            found: true,
            user: {
              id: user.id,
              username: user.username,
              firstName: user.first_name,
              lastName: user.last_name,
              phone: user.phone,
              avatar: user.avatar_url,
              online: onlineUsers.has(user.id),
            },
          })
        } else {
          callback({ found: false })
        }
      } catch {
        callback({ found: false, error: "Lookup failed" })
      }
    })

    socket.on("call:offer", async ({ to, sdp, callerInfo }) => {
      const targetSocketId = onlineUsers.get(to)
      if (targetSocketId) {
        let info = callerInfo
        if (!info) {
          try {
            const result = await pool.query(
              `SELECT id, username, first_name, last_name, avatar_url
               FROM users WHERE id = $1`,
              [userId],
            )
            if (result.rows.length > 0) {
              const u = result.rows[0]
              info = {
                id: u.id,
                username: u.username,
                firstName: u.first_name,
                lastName: u.last_name,
                avatar: u.avatar_url,
              }
            }
          } catch {}
        }
        io.to(targetSocketId).emit("call:incoming", {
          from: userId,
          sdp,
          callerInfo: info,
        })
      } else {
        socket.emit("call:user-offline", { userId: to })
      }
    })

    socket.on("call:answer", ({ to, sdp }) => {
      const targetSocketId = onlineUsers.get(to)
      if (targetSocketId) {
        io.to(targetSocketId).emit("call:answered", { from: userId, sdp })
      }
    })

    socket.on("call:ice-candidate", ({ to, candidate }) => {
      const targetSocketId = onlineUsers.get(to)
      if (targetSocketId) {
        io.to(targetSocketId).emit("call:ice-candidate", { from: userId, candidate })
      }
    })

    socket.on("call:end", ({ to }) => {
      const targetSocketId = onlineUsers.get(to)
      if (targetSocketId) {
        io.to(targetSocketId).emit("call:ended", { from: userId })
      }
    })

    socket.on("call:reject", ({ to }) => {
      const targetSocketId = onlineUsers.get(to)
      if (targetSocketId) {
        io.to(targetSocketId).emit("call:rejected", { from: userId })
      }
    })

    socket.on("call:busy", ({ to }) => {
      const targetSocketId = onlineUsers.get(to)
      if (targetSocketId) {
        io.to(targetSocketId).emit("call:busy", { from: userId })
      }
    })

    socket.on("message:deleted", async ({ conversationId, messageId, senderId }) => {
      try {
        const result = await pool.query(
          `SELECT cp.user_id FROM conversation_participants cp
           WHERE cp.conversation_id = $1 AND cp.user_id != $2`,
          [conversationId, senderId || userId],
        )
        for (const row of result.rows) {
          const targetSocketId = onlineUsers.get(row.user_id)
          if (targetSocketId) {
            io.to(targetSocketId).emit("message:deleted", { conversationId, messageId })
          }
        }
        console.log(`[message:deleted] broadcast msg=${messageId} conv=${conversationId} by=${senderId || userId}`)
      } catch {
        console.warn(`[message:deleted] failed to broadcast msg=${messageId}`)
      }
    })
  }

  socket.on("room:join", ({ roomId, displayName }) => {
    const participantId = userId || `anon-${socket.id}`
    const participantLabel = displayName || participantId

    if (!rooms.has(roomId)) {
      rooms.set(roomId, [])
    }
    const participants = rooms.get(roomId)
    participants.push({ socketId: socket.id, id: participantId, label: participantLabel })
    socket.join(roomId)

    const list = participants.map(p => ({ id: p.id, label: p.label }))

    if (participants.length === 1) {
      socket.emit("room:waiting", { roomId, participants: list })
    } else if (participants.length === 2) {
      io.to(participants[0].socketId).emit("room:initiate-offer", { roomId, participants: list })
      io.to(participants[1].socketId).emit("room:participant-joined", { roomId, participants: list })
    } else {
      socket.emit("room:participant-joined", { roomId, participants: list })
      socket.to(roomId).emit("room:participant-joined", { roomId, participants: list })
    }
  })

  socket.on("room:offer", ({ roomId, sdp }) => {
    socket.to(roomId).emit("room:offer", { sdp })
  })

  socket.on("room:answer", ({ roomId, sdp }) => {
    socket.to(roomId).emit("room:answer", { sdp })
  })

  socket.on("room:ice-candidate", ({ roomId, candidate }) => {
    socket.to(roomId).emit("room:ice-candidate", { candidate })
  })

  socket.on("room:leave", ({ roomId }) => {
    socket.leave(roomId)
    const participants = rooms.get(roomId)
    if (participants) {
      const idx = participants.findIndex(p => p.socketId === socket.id)
      if (idx !== -1) participants.splice(idx, 1)
      if (participants.length === 0) {
        rooms.delete(roomId)
      } else {
        const list = participants.map(p => ({ id: p.id, label: p.label }))
        io.to(roomId).emit("room:participant-left", { roomId, participants: list })
      }
    }
  })

  socket.on("disconnect", () => {
    for (const [roomId, participants] of rooms) {
      const idx = participants.findIndex(p => p.socketId === socket.id)
      if (idx !== -1) {
        participants.splice(idx, 1)
        if (participants.length === 0) {
          rooms.delete(roomId)
        } else {
          const list = participants.map(p => ({ id: p.id, label: p.label }))
          io.to(roomId).emit("room:participant-left", { roomId, participants: list })
        }
      }
    }
    if (userId) {
      onlineUsers.delete(userId)
      socket.broadcast.emit("user:offline", { userId })
    }
  })
})

httpServer.listen(PORT, () => {
  console.log(`[signaling] server running on port ${PORT}`)
})