Public/apps/im-server/server.js

#!/usr/bin/env node
/**
 * MobileIMSDK TCP 服务端（官方协议）：
 * 4 字节大端长度 + UTF-8 JSON Protocal。
 * 登录 token 走 OA GET /auth/me。
 * 同一账号允许多端同时在线（手机 + 网页），重连不再把本机踢掉。
 */
const net = require('net');
const http = require('http');
const crypto = require('crypto');

const TCP_PORT = Number(process.env.IM_TCP_PORT || 8901);
const HTTP_PORT = Number(process.env.IM_HTTP_PORT || 8902);
const OA_API = (process.env.OA_API_URL || 'http://127.0.0.1:3010/api/v1').replace(/\/$/, '');
const INTERNAL = process.env.IM_INTERNAL_TOKEN || '';
const MAX_BODY = 6 * 1024;

const C = {
  LOGIN: 0,
  KEEP_ALIVE: 1,
  COMMON_DATA: 2,
  LOGOUT: 3,
  RECIVED: 4,
  ECHO: 5,
};
const S = {
  LOGIN: 50,
  KEEP_ALIVE: 51,
  ERROR: 52,
  ECHO: 53,
  KICKOUT: 54,
};

/** @type {Map<string, Set<import('net').Socket>>} */
const online = new Map();
/** @type {WeakMap<import('net').Socket, { userId?: string, buf: Buffer }>} */
const sessions = new WeakMap();

function frame(obj) {
  const body = Buffer.from(JSON.stringify(obj), 'utf8');
  const header = Buffer.alloc(4);
  header.writeUInt32BE(body.length, 0);
  return Buffer.concat([header, body]);
}

function send(socket, obj) {
  if (!socket || socket.destroyed) return false;
  try {
    socket.write(frame(obj));
    return true;
  } catch {
    return false;
  }
}

function addOnline(userId, socket) {
  let set = online.get(userId);
  if (!set) {
    set = new Set();
    online.set(userId, set);
  }
  set.add(socket);
}

function removeSocket(userId, socket) {
  const set = online.get(userId);
  if (!set) return;
  set.delete(socket);
  if (set.size === 0) online.delete(userId);
}

function sendToUser(userId, obj) {
  const set = online.get(userId);
  if (!set) return false;
  let ok = false;
  for (const s of set) ok = send(s, obj) || ok;
  return ok;
}

function protocal(type, dataContent, from, to, extra = {}) {
  return {
    bridge: false,
    type,
    dataContent,
    from: from || '0',
    to: to || '0',
    fp: extra.fp || null,
    QoS: Boolean(extra.QoS),
    typeu: extra.typeu ?? -1,
    sm: extra.sm ?? -1,
  };
}

async function verifyLogin(userId, token) {
  if (!userId || !token) return false;
  try {
    const res = await fetch(`${OA_API}/auth/me`, {
      headers: { Authorization: `Bearer ${token}` },
    });
    if (!res.ok) return false;
    const json = await res.json();
    const me = json.data || json;
    return me && me.id === userId && me.status !== 'LEFT';
  } catch (e) {
    console.warn('verify login', e.message);
    return false;
  }
}

async function ingest(fromUserId, toUserId, body, fingerprint, extra = {}) {
  if (!INTERNAL) return;
  try {
    await fetch(`${OA_API}/im/internal/ingest`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json', 'x-im-token': INTERNAL },
      body: JSON.stringify({
        fromUserId,
        toUserId,
        kind: extra.kind || 'chat',
        body,
        fingerprint,
        conversationId: extra.conversationId,
        contentType: extra.contentType,
        meta: extra.meta,
      }),
    });
  } catch (e) {
    console.warn('ingest', e.message);
  }
}

async function membersOf(conversationId) {
  if (!INTERNAL || !conversationId) return [];
  try {
    const res = await fetch(`${OA_API}/im/internal/members/${conversationId}`, {
      headers: { 'x-im-token': INTERNAL },
    });
    if (!res.ok) return [];
    const json = await res.json();
    const data = json.data || json;
    return data.userIds || [];
  } catch {
    return [];
  }
}

async function handle(socket, p) {
  const type = Number(p.type);
  if (type === C.LOGIN) {
    let info = {};
    try {
      info = JSON.parse(p.dataContent || '{}');
    } catch {
      info = {};
    }
    const userId = info.loginUserId || p.from;
    const ok = await verifyLogin(userId, info.loginToken);
    if (!ok) {
      send(socket, protocal(S.LOGIN, JSON.stringify({ code: 1, firstLoginTime: -1 }), '0', '-1'));
      socket.end();
      return;
    }
    const firstLoginTime = Number(info.firstLoginTime) > 0 ? Number(info.firstLoginTime) : Date.now();
    const sess = sessions.get(socket) || { buf: Buffer.alloc(0) };
    sess.userId = userId;
    sessions.set(socket, sess);
    addOnline(userId, socket);
    send(socket, protocal(S.LOGIN, JSON.stringify({ code: 0, firstLoginTime }), '0', userId));
    console.log(`login ok ${userId} sockets=${online.get(userId)?.size || 0} users=${online.size}`);
    return;
  }

  const sess = sessions.get(socket);
  if (!sess?.userId) {
    socket.end();
    return;
  }
  const userId = sess.userId;

  if (type === C.KEEP_ALIVE) {
    send(socket, protocal(S.KEEP_ALIVE, '{}', '0', userId));
    return;
  }
  if (type === C.LOGOUT) {
    removeSocket(userId, socket);
    socket.end();
    return;
  }
  if (type === C.ECHO) {
    send(socket, protocal(S.ECHO, p.dataContent || '', '0', userId));
    return;
  }
  if (type === C.RECIVED) {
    return;
  }
  if (type === C.COMMON_DATA) {
    const fp = p.fp || crypto.randomUUID();
    let parsed = {};
    try {
      parsed = JSON.parse(p.dataContent || '{}');
    } catch {
      parsed = {};
    }
    const convId = parsed.conversationId || '';
    const typeu = Number(p.typeu ?? -1);
    const payload = protocal(C.COMMON_DATA, p.dataContent, userId, p.to, {
      QoS: p.QoS,
      fp,
      typeu: p.typeu,
      sm: Date.now(),
    });
    let targets = [];
    if (typeu === 2 && convId) {
      targets = (await membersOf(convId)).filter((id) => id !== userId);
    } else if (p.to) {
      targets = [p.to];
    }
    for (const to of targets) {
      sendToUser(to, { ...payload, to });
    }
    if (p.QoS) {
      send(socket, protocal(C.RECIVED, fp, p.to || '0', userId));
    }
    if (typeu === 3 || parsed.kind === 'signal') return;
    const text = String(parsed.text || parsed.title || p.dataContent || '');
    await ingest(userId, p.to, text, fp, {
      kind: parsed.kind || 'chat',
      conversationId: convId,
      contentType: parsed.contentType,
      meta: parsed.meta,
    });
  }
}

const tcp = net.createServer((socket) => {
  sessions.set(socket, { buf: Buffer.alloc(0) });
  socket.on('data', (chunk) => {
    const sess = sessions.get(socket);
    if (!sess) return;
    sess.buf = Buffer.concat([sess.buf, chunk]);
    while (sess.buf.length >= 4) {
      const len = sess.buf.readUInt32BE(0);
      if (len <= 0 || len > MAX_BODY) {
        socket.end();
        return;
      }
      if (sess.buf.length < 4 + len) break;
      const body = sess.buf.subarray(4, 4 + len).toString('utf8');
      sess.buf = sess.buf.subarray(4 + len);
      try {
        void handle(socket, JSON.parse(body));
      } catch (e) {
        console.warn('bad packet', e.message);
      }
    }
  });
  socket.on('close', () => {
    const sess = sessions.get(socket);
    if (sess?.userId) removeSocket(sess.userId, socket);
  });
  socket.on('error', () => {});
  socket.setTimeout(90000, () => socket.end());
});

const httpServer = http.createServer((req, res) => {
  const url = new URL(req.url || '/', 'http://127.0.0.1');
  if (req.method === 'GET' && url.pathname === '/health') {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ ok: true, online: online.size, tcp: TCP_PORT }));
    return;
  }
  if (req.method === 'GET' && url.pathname === '/online') {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ count: online.size, users: [...online.keys()] }));
    return;
  }
  if (req.method === 'POST' && url.pathname === '/push') {
    const token = req.headers['x-im-token'];
    if (!INTERNAL || token !== INTERNAL) {
      res.writeHead(401);
      res.end('forbidden');
      return;
    }
    const chunks = [];
    req.on('data', (c) => chunks.push(c));
    req.on('end', () => {
      try {
        const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
        const to = body.toUserId;
        let parsed = {};
        try {
          parsed = JSON.parse(body.dataContent || '{}');
        } catch {
          parsed = {};
        }
        const seq = Number(parsed.seq ?? body.seq ?? 0);
        const delivered = sendToUser(
          to,
          protocal(C.COMMON_DATA, body.dataContent || '', '0', to, {
            QoS: true,
            fp: parsed.fingerprint || crypto.randomUUID(),
            typeu: 1,
            sm: seq > 0 ? seq : Date.now(),
          }),
        );
        res.writeHead(200, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ ok: true, delivered }));
      } catch (e) {
        res.writeHead(400);
        res.end(e.message);
      }
    });
    return;
  }
  res.writeHead(404);
  res.end();
});

tcp.listen(TCP_PORT, '0.0.0.0', () => {
  console.log(`MobileIMSDK TCP ${TCP_PORT}`);
});
httpServer.listen(HTTP_PORT, '127.0.0.1', () => {
  console.log(`IM HTTP push ${HTTP_PORT}`);
});