yuuza/outline
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'tom/socket-refactor'

Browse Source
This commit is contained in:
Tom Moor
2022-08-27 11:51:38 +02:00
parent 4dbad4e46c 354a68a8b7
commit 1c8fadbe02
43 changed files with 624 additions and 719 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
server/queues/processors/WebsocketsProcessor.ts
View File

@@ -10,11 +10,16 @@ import {
GroupUser,
Pin,
Star,
Team,
} from "@server/models";
import {
presentCollection,
presentDocument,
presentFileOperation,
presentGroup,
presentPin,
presentStar,
presentTeam,
} from "@server/presenters";
import { Event } from "../../types";
@@ -23,7 +28,6 @@ export default class WebsocketsProcessor {
switch (event.name) {
case "documents.publish":
case "documents.restore":
case "documents.archive":
case "documents.unarchive": {
const document = await Document.findByPk(event.documentId, {
paranoid: false,
@@ -51,52 +55,16 @@ export default class WebsocketsProcessor {
});
}
case "documents.delete": {
const document = await Document.findByPk(event.documentId, {
paranoid: false,
});
if (!document) {
return;
}
if (!document.publishedAt) {
return socketio.to(`user-${document.createdById}`).emit("entities", {
event: event.name,
documentIds: [
{
id: document.id,
updatedAt: document.updatedAt,
},
],
});
}
return socketio
.to(`collection-${document.collectionId}`)
.emit("entities", {
event: event.name,
documentIds: [
{
id: document.id,
updatedAt: document.updatedAt,
},
],
collectionIds: [
{
id: document.collectionId,
},
],
});
}
case "documents.permanent_delete": {
return socketio
.to(`collection-${event.collectionId}`)
.emit(event.name, {
documentId: event.documentId,
modelId: event.documentId,
});
}
case "documents.archive":
case "documents.delete":
case "documents.update": {
const document = await Document.findByPk(event.documentId, {
paranoid: false,
@@ -107,15 +75,9 @@ export default class WebsocketsProcessor {
const channel = document.publishedAt
? `collection-${document.collectionId}`
: `user-${event.actorId}`;
return socketio.to(channel).emit("entities", {
event: event.name,
documentIds: [
{
id: document.id,
updatedAt: document.updatedAt,
},
],
});
const data = await presentDocument(document);
return socketio.to(channel).emit(event.name, data);
}
case "documents.create": {
@@ -139,13 +101,6 @@ export default class WebsocketsProcessor {
});
}
case "documents.star":
case "documents.unstar": {
return socketio.to(`user-${event.actorId}`).emit(event.name, {
documentId: event.documentId,
});
}
case "documents.move": {
const documents = await Document.findAll({
where: {
@@ -188,22 +143,15 @@ export default class WebsocketsProcessor {
.to(
collection.permission
? `team-${collection.teamId}`
: `collection-${collection.id}`
: `user-${collection.createdById}`
)
.emit("entities", {
event: event.name,
collectionIds: [
{
id: collection.id,
updatedAt: collection.updatedAt,
},
],
});
.emit(event.name, presentCollection(collection));
return socketio
.to(
collection.permission
? `team-${collection.teamId}`
: `collection-${collection.id}`
: `user-${collection.createdById}`
)
.emit("join", {
event: event.name,
@@ -211,8 +159,7 @@ export default class WebsocketsProcessor {
});
}
case "collections.update":
case "collections.delete": {
case "collections.update": {
const collection = await Collection.findByPk(event.collectionId, {
paranoid: false,
});
@@ -230,6 +177,14 @@ export default class WebsocketsProcessor {
});
}
case "collections.delete": {
return socketio
.to(`collection-${event.collectionId}`)
.emit(event.name, {
modelId: event.collectionId,
});
}
case "collections.move": {
return socketio
.to(`collection-${event.collectionId}`)
@@ -366,8 +321,9 @@ export default class WebsocketsProcessor {
if (!fileOperation) {
return;
}
const data = await presentFileOperation(fileOperation);
return socketio.to(`user-${event.actorId}`).emit(event.name, data);
return socketio
.to(`user-${event.actorId}`)
.emit(event.name, presentFileOperation(fileOperation));
}
case "pins.create":
@@ -422,15 +378,9 @@ export default class WebsocketsProcessor {
if (!group) {
return;
}
return socketio.to(`team-${group.teamId}`).emit("entities", {
event: event.name,
groupIds: [
{
id: group.id,
updatedAt: group.updatedAt,
},
],
});
return socketio
.to(`team-${group.teamId}`)
.emit(event.name, presentGroup(group));
}
case "groups.add_user": {
@@ -518,25 +468,14 @@ export default class WebsocketsProcessor {
}
case "groups.delete": {
const group = await Group.findByPk(event.modelId, {
paranoid: false,
socketio.to(`team-${event.teamId}`).emit(event.name, {
modelId: event.modelId,
});
if (!group) {
return;
}
socketio.to(`team-${group.teamId}`).emit("entities", {
event: event.name,
groupIds: [
{
id: group.id,
updatedAt: group.updatedAt,
},
],
});
// we the users and collection relations that were just severed as a result of the group deletion
// since there are cascading deletes, we approximate this by looking for the recently deleted
// items in the GroupUser and CollectionGroup tables
// we get users and collection relations that were just severed as a
// result of the group deletion since there are cascading deletes, we
// approximate this by looking for the recently deleted items in the
// GroupUser and CollectionGroup tables
const groupUsers = await GroupUser.findAll({
paranoid: false,
where: {
@@ -594,14 +533,13 @@ export default class WebsocketsProcessor {
}
case "teams.update": {
return socketio.to(`team-${event.teamId}`).emit("entities", {
event: event.name,
teamIds: [
{
id: event.teamId,
},
],
});
const team = await Team.scope("withDomains").findByPk(event.teamId);
if (!team) {
return;
}
return socketio
.to(`team-${event.teamId}`)
.emit(event.name, presentTeam(team));
}
default:

2
server/queues/tasks/DeliverWebhookTask.ts
View File

@@ -122,8 +122,6 @@ export default class DeliverWebhookTask extends BaseTask<Props> {
case "documents.archive":
case "documents.unarchive":
case "documents.restore":
case "documents.star":
case "documents.unstar":
case "documents.move":
case "documents.update":
case "documents.title_change":

18
server/routes/api/__snapshots__/documents.test.ts.snap
View File

@@ -44,24 +44,6 @@ Object {
}
`;
exports[`#documents.star should require authentication 1`] = `
Object {
"error": "authentication_required",
"message": "Authentication required",
"ok": false,
"status": 401,
}
`;
exports[`#documents.unstar should require authentication 1`] = `
Object {
"error": "authentication_required",
"message": "Authentication required",
"ok": false,
"status": 401,
}
`;
exports[`#documents.update should fail if document lastRevision does not match 1`] = `
Object {
"error": "invalid_request",

74
server/routes/api/documents.test.ts
View File

@@ -1,7 +1,6 @@
import {
Document,
View,
Star,
Revision,
Backlink,
CollectionUser,
@@ -1819,79 +1818,6 @@ describe("#documents.restore", () => {
});
});
describe("#documents.star", () => {
it("should star the document", async () => {
const { user, document } = await seed();
const res = await server.post("/api/documents.star", {
body: {
token: user.getJwtToken(),
id: document.id,
},
});
const stars = await Star.findAll();
expect(res.status).toEqual(200);
expect(stars.length).toEqual(1);
expect(stars[0].documentId).toEqual(document.id);
});
it("should require authentication", async () => {
const res = await server.post("/api/documents.star");
const body = await res.json();
expect(res.status).toEqual(401);
expect(body).toMatchSnapshot();
});
it("should require authorization", async () => {
const { document } = await seed();
const user = await buildUser();
const res = await server.post("/api/documents.star", {
body: {
token: user.getJwtToken(),
id: document.id,
},
});
expect(res.status).toEqual(403);
});
});
describe("#documents.unstar", () => {
it("should unstar the document", async () => {
const { user, document } = await seed();
await Star.create({
documentId: document.id,
userId: user.id,
});
const res = await server.post("/api/documents.unstar", {
body: {
token: user.getJwtToken(),
id: document.id,
},
});
const stars = await Star.findAll();
expect(res.status).toEqual(200);
expect(stars.length).toEqual(0);
});
it("should require authentication", async () => {
const res = await server.post("/api/documents.star");
const body = await res.json();
expect(res.status).toEqual(401);
expect(body).toMatchSnapshot();
});
it("should require authorization", async () => {
const { document } = await seed();
const user = await buildUser();
const res = await server.post("/api/documents.unstar", {
body: {
token: user.getJwtToken(),
id: document.id,
},
});
expect(res.status).toEqual(403);
});
});
describe("#documents.import", () => {
it("should error if no file is passed", async () => {
const user = await buildUser();

70
server/routes/api/documents.ts
View File

@@ -23,7 +23,6 @@ import {
Event,
Revision,
SearchQuery,
Star,
User,
View,
} from "@server/models";
@@ -731,75 +730,6 @@ router.post(
}
);
// Deprecated – use stars.create instead
router.post("documents.star", auth(), async (ctx) => {
const { id } = ctx.body;
assertPresent(id, "id is required");
const { user } = ctx.state;
const document = await Document.findByPk(id, {
userId: user.id,
});
authorize(user, "read", document);
await Star.findOrCreate({
where: {
documentId: document.id,
userId: user.id,
},
});
await Event.create({
name: "documents.star",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: user.id,
data: {
title: document.title,
},
ip: ctx.request.ip,
});
ctx.body = {
success: true,
};
});
// Deprecated – use stars.delete instead
router.post("documents.unstar", auth(), async (ctx) => {
const { id } = ctx.body;
assertPresent(id, "id is required");
const { user } = ctx.state;
const document = await Document.findByPk(id, {
userId: user.id,
});
authorize(user, "read", document);
await Star.destroy({
where: {
documentId: document.id,
userId: user.id,
},
});
await Event.create({
name: "documents.unstar",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: user.id,
data: {
title: document.title,
},
ip: ctx.request.ip,
});
ctx.body = {
success: true,
};
});
router.post("documents.templatize", auth({ member: true }), async (ctx) => {
const { id } = ctx.body;
assertPresent(id, "id is required");

25
server/services/websockets.ts
View File

@@ -6,6 +6,8 @@ import IO from "socket.io";
import { createAdapter } from "socket.io-redis";
import Logger from "@server/logging/Logger";
import Metrics from "@server/logging/metrics";
import * as Tracing from "@server/logging/tracing";
import { APM } from "@server/logging/tracing";
import { Document, Collection, View, User } from "@server/models";
import { can } from "@server/policies";
import { getUserForJWT } from "@server/utils/jwt";
@@ -135,14 +137,23 @@ export default function init(
// Handle events from event queue that should be sent to the clients down ws
const websockets = new WebsocketsProcessor();
websocketQueue.process(async function websocketEventsProcessor(job) {
const event = job.data;
websockets.perform(event, io).catch((error) => {
Logger.error("Error processing websocket event", error, {
event,
websocketQueue.process(
APM.traceFunction({
serviceName: "websockets",
spanName: "process",
isRoot: true,
})(async function (job) {
const event = job.data;
Tracing.setResource(`Processor.WebsocketsProcessor`);
websockets.perform(event, io).catch((error) => {
Logger.error("Error processing websocket event", error, {
event,
});
});
});
});
})
);
}
async function authenticated(io: IO.Server, socket: SocketWithAuth) {

4
server/types.ts
View File

@@ -95,9 +95,7 @@ export type DocumentEvent = BaseEvent &
| "documents.permanent_delete"
| "documents.archive"
| "documents.unarchive"
| "documents.restore"
| "documents.star"
| "documents.unstar";
| "documents.restore";
documentId: string;
collectionId: string;
data: {