import nedb from 'nedb' import express from 'express' import expressWs from 'express-ws' import session from 'express-session' import sessionDb from 'express-session-nedb' import random from 'string-random' import formidable from 'formidable' import crypto from 'crypto' import HUB from './fmhub.js' import { createServer } from 'vite' const md5 = (str) => crypto.createHash('md5').update(str).digest('hex') const databases = new Map() // 所有数据库 const FM = new HUB() // 频道消息分发器 const messagelist = new Map() // 消息队列的nedb存储 // 服务端推送 // 1. 用户消息 // 2. 系统消息 // 3. 聊天消息 // 4. 资源变化(观看中的资源变化)(预取集, 观看集, 缓存集) // 5. 操作同步(其它用户的协作动作)(用户是临时附着与资源的对象, 由于其变化频繁且可丢弃, 通过会话通道读取) const db = (name) => (databases.get(name) || function () { let database = new nedb({ filename: `./data/db/${name}.db`, autoload: true, timestampData: true }) databases.set(name, database) return database }()) const session_store = sessionDb(session, db('session')) // 登录验证 const online = function (req, res, next) { if (!req.session.account) return res.status(401).send('未登录') if (req.session.account.gid !== 1) req.params.uid = req.session.account.uid // 普通用户只能操作自己的 next() } // 列表计量 const count_load = async (name, query) => await new Promise(resolve => db(name).count(query, (err, count) => resolve(count))) // 条件查询 const list_load = async (name, query) => await new Promise(resolve => db(name).find(query, function (err, docs) { return resolve(docs.Map((item, index) => Object.assign({}, { _id: item.id }))) })) const user_load = async (_id) => await new Promise(resolve => db('user').findOne({ _id }, function (err, doc) { if (!doc) return resolve(doc) // let { salt, password, mobile, email, ...user } = doc // 这里只应提取有限简略信息附给列表, 因为个人信息中可能含有大量私有字段 // 但是..nedb 全在内存, 效率略略略... let { _id, gid, name, avatar } = doc return resolve({ _id, gid, name, avatar }) })) // 通讯频道 Frequency Modulation, 游客使用公共账户 uid = 0 function websocketer(ws, req) { let uid = req.session?.account?.uid || "0" console.log(`用户 ${uid} 连接了服务器`) FM.增加会话(uid, ws) ws.on('message', function (msg) { if (typeof (msg) !== "string") return console.log("消息不是字符串") let { fm, data } = JSON.parse(msg) // 消息不是JSON FM.发送消息(fm, uid, data) }) ws.on('close', (code) => FM.移除会话(uid, ws)) ws.on('error', (code) => console.log('link error: ', code)) } // 会话列表 function session_list(req, res) { return session_store.db.find({ "data.account.uid": req.session.account.uid }, function (err, docs) { return err ? res.status(500).send('错误') : res.json(docs) }) } import Cache from './src/cache.js' const cache = new Cache() // 登录会话 function session_create(req, res) { let { mobile, code, name, password } = req.body // 也许使用验证码登录(临时验证码) && code !== '000000' if (mobile && code) { if (code !== cache.get(mobile)) return res.status(400).send('验证码错误') return db('user').findOne({ mobile }, function (err, doc) { if (!doc) return res.status(400).send('账户不存在') return req.session.regenerate(function (err) { req.session.account = { uid: doc._id, gid: doc.gid ?? 0 } req.session.isAdmin = () => (req.session.account.gid === 1) // 是否管理员 let { salt, password, ...user } = doc return res.json(user) }) }) } // 也许使用账号密码登录 return db('user').findOne({ name }, function (err, doc) { if (!doc) return res.status(400).send('账户不存在') if (md5(password + doc.salt) !== doc.password) return res.status(400).send('密码错误') return req.session.regenerate(function (err) { req.session.account = { uid: doc._id, gid: doc.gid ?? 0 } let { salt, password, ...user } = doc return res.json(user) }) }) } // 注销会话 (当前会话) function sessionDeleteSelf(req, res) { return req.session.destroy(function (err) { return res.status(err ? 500 : 200).send(err ? '错误' : '退出登录') }) } // 注销会话 (指定会话) function session_delete(req, res) { return req.sessionStore.destroy(req.params.sid, function (err) { return err ? res.status(500).send('错误') : res.send('退出登录') }) } // 账户资料 (当前账户) function profile(req, res) { if (!req.session.account) return res.json({ name: '游客', avatar: '', online: false }) return db('user').findOne({ _id: req.session.account.uid }, function (err, doc) { if (err) return res.status(400).send('账户不存在') doc.online = true delete doc.salt delete doc.password return res.json(doc) }) } // 列表对象 const object_list = async function (req, res) { let { pagesize, page, count, like, post, tid, top, uid, user, sort, desc, ...query } = req.query if (tid) query.tid = Number(tid) // 某些查询参数需要转换类型 if (top) query.top = Number(top) // 某些查询参数需要转换类型 if (uid && uid !== req.session?.account?.uid) query.public = true // 如果查询条件限定为自己的, 则不用限制范围到公开的 page = Number(page) || 1 // 默认页码1 pagesize = Number(pagesize) || 20 // 默认分页20 let skip = (page - 1) * pagesize // 截取点 // 登录状态时, 查询自己点赞过的和评论过的 if (req.session?.account?.uid) { if (like) query.$or = await list_load('like', { attach: req.params.name, uid: req.session.account.uid }) if (post) query.$or = await list_load('post', { attach: req.params.name, uid: req.session.account.uid }) } // 消息限定范围 if (req.params.name === 'message' && req.session.account.gid !== 1) { if (query.to) query.to = req.session.account.uid // 只能查发给自己的消息 else query.uid = req.session.account.uid // 否则默认自己发出去的 } // 要求附带统计信息 if (count) await new Promise(resolve => db(req.params.name).count(query, function (err, count) { res.header('count', count) res.header('page', page) res.header('pages', Math.ceil(count / pagesize)) res.header('pagesize', pagesize) resolve() })) desc = (desc === "1") ? 1 : -1 let is_sort = {} switch (sort) { case 'top': is_sort.top = desc; break; case 'hot': is_sort.hot = desc; break; case 'createdAt': is_sort.createdAt = desc; break; case 'updatedAt': is_sort.updatedAt = desc; break; default: } return db(req.params.name).find(query).skip(skip).limit(pagesize).sort(sort).exec(async function (err, docs) { return res.json(await Promise.all(docs.map(async item => { item.posts = await count_load('post', { attach: req.params.name, aid: item._id }) // 附加评论数量 item.likes = await count_load('like', { attach: req.params.name, aid: item._id }) // 附加点赞数量 if (req.params.name === 'user') { delete item.salt delete item.password delete item.mobile delete item.email } else { item.user = await user_load(item.uid) // 附加用户信息(user对象没有作者) } if (req.params.name != 'user' && req.session.account) { item.like = !!(await count_load('like', { attach: req.params.name, aid: item._id, uid: req.session.account.uid })) } return item }))) }) } const isAdmin = (req) => req.session?.account?.gid === 1 const 权限模组 = (req, res, next) => { // 创建任何对象, 如果不是管理员, 则不能设置 gid 和计算属性以及默认属性 if (req.session?.account?.gid !== 1) { ['_id', 'uid', 'top', 'user', 'createdAt', 'updatedAt', 'views', 'posts', 'likes', 'files'].forEach(i => delete req.body[i]) } } // 创建对象 const object_create = async function (req, res) { // 游客和普通用户禁止设置, 权限和计算属性 if (req.session?.account?.gid != 1) { ['_id', 'uid', 'top', 'user', 'createdAt', 'updatedAt', 'views', 'posts', 'likes', 'files'].forEach(i => delete req.body[i]) } // 如果创建对象是用户作一些特殊处理, 否则创建对象作通用处理 if (req.params.name === 'user') { const { name, password, avatar = '' } = req.body if (!name) return res.status(400).send('用户名不能为空') if (!password) return res.status(400).send('密码不能为空') if (await count_load('user', { name })) return res.status(400).send('用户名已被占用') req.body = { ...req.body, avatar, gid: (await count_load('user', {})) ? 0 : 1, // 默认是管理员为首个注册用户 salt: random(32), // 密码加盐 password: md5(password + random(32)) // 必要设置密码 } } else { if (!req.session.account) return res.status(401).send('需要登录') req.body = { ...req.body, uid: req.session.account.uid, public: true, views: 0 } } // 如果包含标签(先查询是否存在, 存在则使用返回的_id进行挂载, 不存在则创建新的) if (req.body.tags && Array.isArray(req.body.tags)) { req.body.tags.forEach(item => { db('tag').findOne({ name: item }, function (err, doc) { if (err && !doc) { return // 创建新的 } else { return // 使用这个 _id, 向它写入 } }) }) // 是否可以创建一个复杂关系型数据库? // 以应对映射的共同对象 // 例如在使用 tag 时, 向 idea 表的 tag 段读写, 即是 tag表的 idea 索引范围 // (自动构建和维护双向索引) // 当删除此 idea 时, 也自动清理掉 tag 对 idea 的连接 } // 如果是挂载对象到指定目标 if (req.body.attach && req.body.aid) { let count = await count_load(req.body.attach, { _id: req.body.aid }) if (!count) return res.status(404).send('目标挂载对象不存在') } // 写入对象 return db(req.params.name).insert(req.body, async function (err, doc) { if (!doc) return res.status(500).send('创建失败') if (req.params.name !== 'user') { doc.user = await user_load(doc.uid) } else { delete doc.salt delete doc.password } return res.json(doc) }) } const 密码加密 = (password, salt = random(32)) => ({ salt, password: md5(password + salt) }) const 密码解密 = (password, salt) => md5(password + salt) // 修改对象 function object_patch(req, res, next) { return db(req.params.name).findOne({ _id: req.params._id }, async function (err, doc) { if (!doc) return res.status(404).send('目标对象不存在') if (typeof (req.body.attach) !== "undefined") return res.status(403).send("无权限修改挂载目标") if (typeof (req.body.aid) !== "undefined") return res.status(403).send("无权限修改挂载目标") // 如果是 user 做一些特殊处理 if (req.params.name === 'user') { if (req.session.account.gid !== 1) { if (req.session.account.uid !== doc._id) return res.status(403).send('没有权限修改账户') if (typeOf(req.body.gid) == "undefined") return res.status(403).send('没有权限修改权限') } if (req.body.password) { req.body = { ...req.body, ...密码加密(req.body.password) } } if (req.body.name && await count_load('user', { name: req.body.name })) return res.status(400).send('用户名已被占用') } else { if (req.session.account.uid !== doc.uid && req.session.account.gid !== 1) return res.status(403).send('没有权限修改对象') if (req.body.uid && req.session.account.gid !== 1) return res.status(403).send('没有权限修改归属') } return db(req.params.name).update({ _id: req.params._id }, { $set: req.body }, function (err, count) { if (!count) return res.status(500).send('修改失败') res.send('修改成功') // 会话完全成功后执行 }) }) } // 删除对象 const object_remove = function (req, res) { return db(req.params.name).findOne({ _id: req.params._id }, async function (err, doc) { if (!doc) return res.status(404).send('目标对象不存在') // 如果是删除用户作一些特殊处理 if (req.params.name === 'user') { if (req.session.account.gid !== 1 && req.session.account.uid !== doc._id) { return res.status(400).send('没有权限删除此账户') } if (await count_load('account', { _id: req.params.id, gid: 1 }) === 1) { return res.status(400).send('不可以删除唯一的管理员账户') } } else { if (req.session.account.gid !== 1 && req.session.account.uid !== doc.uid) { return res.status(403).send('没有权限删除此对象') } } // TODO: 处理掉一些附属对象 // 似乎要遍历所有对象类型? return db(req.params.name).remove({ _id: req.params._id }, function (err, count) { return count ? res.send('删除成功') : res.status(403).send('删除失败') // TODO: 当对象被删除时通过此连接通知所有在线终端 }) }) } // 读取对象 const object_load = function (req, res) { return db(req.params.name).findOne({ _id: req.params._id }, async function (err, doc) { if (!doc) return res.status(404).send('目标资源不存在') if (!doc.public && doc.uid !== session?.account?.uid) return res.status(403).send('没有权限读取') if (req.params.name === 'user') ['salt', 'password', 'mobile', 'email'].forEach(field => delete doc[field]) else doc.user = await user_load(doc.uid) db(req.params.name).update({ _id: req.params._id }, { $set: { views: (doc.views || 0) + 1 } }) return res.json(doc) }) } // 附件上传 const file_upload = function (req, res) { return db(req.params.name).findOne({ _id: req.params._id }, function (err, doc) { if (!doc) return res.status(404).send('目标对象不存在') if (req.session.account.uid !== doc.uid && req.session.account.gid !== 1) { return res.status(403).send('没有权限上传') } return formidable({ multiples: true, uploadDir: 'data/file', keepExtensions: true, maxFieldsSize: 200 * 1024 * 1024 }).parse(req, function (err, fields, files) { const list = Object.values(files).map(item => Array.isArray(item) ? item : [item]).flat().map(({ filepath, mimetype, mtime, newFilename, originalFilename, size }) => { return { filepath, mimetype, mtime, newFilename, originalFilename, size } }) return db(req.params.name).update({ _id: req.params._id }, { $addToSet: { file: { $each: list } } }, function (err, count) { if (!count) return res.status(500).send('附件挂载对象失败') console.log(list) return res.json(list) }) }) }) } // 向账户上传文件 (头像, 背景, 其它文件) const uploadfile = function (req, res) { let idable = formidable({ multiples: true, uploadDir: 'data/file', keepExtensions: true, maxFieldsSize: 200 * 1024 * 1024, }) idable.parse(req, (err, fields, files) => { let query = { _id: req.session.account.uid } // 针对特定对象处理上传字段(头像) let avatar = files["avatar"] if (avatar) { let list = (Array.isArray(avatar) ? avatar : [avatar]) db('user').update(query, { $addToSet: { file: { $each: list } }, // 保存记录 $set: { avatar: '/data/file/' + list[0].newFilename }, // 替换头像 }) } // 针对特定对象处理上传字段(背景图) let background = files["background"] if (background) { let list = (Array.isArray(background) ? background : [background]) db('user').update(query, { $addToSet: { file: { $each: list } }, // 保存记录 $set: { background: '/data/file/' + list[0].newFilename }, // 替换背景 }) } res.json(files) }) } const db_compact = function (req, res) { db(req.params.name).persistence.compactDatafile() return res.send("ok") } // 读取对象列表 function index_get(req, res) { let list = [] databases.forEach((value, key) => list.push(key)) res.json(list) } const app = expressWs(express()).app const ServerSentEventsClient = {} // Server-Sent Events (发送版本号来决定是否接收更新?) app.use((req, res, next) => { if (req.get('Accept') === 'text/event-stream') { console.log('SSE 连接', req.path) res.set({ 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive' }) res.write('retry: 10000\n\n') res.write(`data: ${JSON.stringify({ fm: 'message', data: 'hello' })}\n\n`) res.write(`data: ${JSON.stringify({ fm: 'message', data: 'hellox' })}\n\n`) if (!ServerSentEventsClient[req.path]) ServerSentEventsClient[req.path] = [] ServerSentEventsClient[req.path].push({ req, res }) req.on('close', () => { console.log('SSE 断开', req.path) let index = ServerSentEventsClient[req.path].findIndex(item => item.res === res) if (index !== -1) ServerSentEventsClient[req.path].splice(index, 1) }) } else { next() } }) // SEE DEMO app.route('/demo').get((req, res) => { res.send(`

SSE Express

`) }) app.use(express.json()) app.use(express.urlencoded({ extended: false })) app.use(session({ secret: 'kana', name: 'sid', resave: false, saveUninitialized: false, cookie: { maxAge: 180 * 24 * 3600000 }, store: session_store })) app.use('/api/data/file/', express.static('data/file')) app.ws('/api', websocketer) app.route('/api/').get(index_get) app.route('/api/account').get(profile).post(online, uploadfile) app.route('/api/session').get(online, session_list).post(session_create).delete(online, sessionDeleteSelf) app.route('/api/session/:sid').delete(online, session_delete) app.route('/api/:name').get(object_list).post(object_create).put(db_compact) app.route('/api/:name/:_id').get(object_load).post(online, file_upload).put().patch(online, object_patch).delete(online, object_remove) // 开发模式下使用 Vite if (process.argv.includes('--dev')) { const vite = await createServer({ server: { middlewareMode: 'html' } }) app.use(vite.middlewares) } else { app.use(express.static('dist')) } app.listen(2333)