Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 18x 18x 18x 18x 18x 18x 18x 1x 1x 1x 17x 1x 1x 1x 16x 16x 16x 16x 16x 1x 1x 1x 15x 15x 14x 14x 13x 13x 13x 1x 1x 1x 1x 1x 1x 1x 14x 14x 14x 14x 14x 71x 71x 71x 71x 8x 8x 8x 8x 8x 8x 8x 4x 4x 4x 4x 4x 4x 2x 2x 2x 8x 4x 4x 4x 3x 1x 1x 4x 4x 4x 4x 2x 2x 2x 8x 8x 8x 6x 6x 6x 6x 6x 6x 6x 6x 6x 6x 2x 2x 4x 4x 4x 4x 2x 2x 2x 6x 2x 2x 2x 2x 2x 4x 4x 4x 4x 4x 4x | import { Server } from "@hocuspocus/server";
import axios from "axios";
import * as Y from "yjs";
import jwt from "jsonwebtoken";
import http from "http";
import dotenv from 'dotenv'
dotenv.config()
// Hostnames
const POSTGRES_HOSTNAME = process.env.POSTGRES_HOSTNAME || "http://api-postgres";
const NEO4J_HOSTNAME = process.env.NEO4J_HOSTNAME || "http://api-neo4j";
// CRDT Server Configuration
const JWT_SECRET_KEY = process.env.JWT_SECRET_KEY;
const CRDT_DEBOUNCE = parseInt(process.env.CRDT_DEBOUNCE) || 5000; // Default debounce time in milliseconds (5 seconds)
const CRDT_MAX_DEBOUNCE = parseInt(process.env.CRDT_MAX_DEBOUNCE) || 30000; // Default maximum debounce time in milliseconds (30 seconds)
Iif (!JWT_SECRET_KEY) {
throw new Error("JWT_SECRET_KEY is not set in the environment variables");
}
const ALGORITHM = "HS256";
const DATA_PERMISSION = [
"READ",
"COMMENT",
"WRITE",
"ADMIN",
"OWNER"
]
const httpAgent = new http.Agent({
keepAlive: true, // Keep the connection alive for reuse
timeout: 30000, // Timeout for idle sockets (30 seconds)
});
const axiosClient = axios.create({
httpAgent: httpAgent
});
// const roomNameRegex = /^(?:postgres\/[0-9A-Fa-f-]{36}\/[0-9A-Fa-f-]{36}|neo4j\/[0-9A-Fa-f-]{36}\/[0-9A-Fa-f-]{36}\/[A-Za-z0-9_-]+)$/;
const roomNameRegex = /^(?:postgres\/[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}\/[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}\/[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}|neo4j_(?:node|link)\/[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}\/[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}\/[A-Za-z0-9_-]+)$/;
// Configure the server …
export const server = new Server({
port: parseInt(process.env.CRDT_PORT || "80"), // Port to listen on
debounce: CRDT_DEBOUNCE, // Debounce time in milliseconds (5 seconds)
maxDebounce: CRDT_MAX_DEBOUNCE, // Maximum debounce time in milliseconds (30 seconds)
// Someone is trying to connect to the server
async onConnect(data) {
console.log("onConnect");
// Output some information
console.log(`New websocket connection`);
// Set as not authenticated by default
data.connectionConfig.isAuthenticated = false; // Set authenticated flag
},
// Someone is trying to authenticate (only if token is provided)
async onAuthenticate(data) {
console.log("onAuthenticate");
const { socketId, documentName, token } = data;
// Output some information
console.log(`New websocket connection from ${socketId} on document ${documentName} with token:`, token);
// if token is empty
if (!token) {
data.connectionConfig.isAuthenticated = false; // Set authenticated flag to false
console.log(`Connection from ${socketId} rejected due to missing token`);
throw new Error("Missing token");
}
// Check if the document name is valid
if (!roomNameRegex.test(documentName)) {
console.log(`Connection from ${socketId} rejected due to invalid document name: ${documentName}`);
data.connectionConfig.isAuthenticated = false; // Set authenticated flag to false
throw new Error("Invalid document name");
}
const documentNameList = documentName.split("/");
const constellation_uuid = documentNameList[1]; // Get the constellation UUID from the document name
// Here we decode the JWT token to get the user information
let decodedToken;
try {
// Verify the token using the secret key and algorithm
decodedToken = jwt.verify(token, JWT_SECRET_KEY, { algorithms: [ALGORITHM] });
// Check if the token has expired
Iif (decodedToken.exp < Math.floor(Date.now() / 1000)) {
console.log("Token has expired");
data.connectionConfig.isAuthenticated = false; // Set authenticated flag to false
throw new Error("Token has expired");
}
} catch (err) {
// If the token is invalid or expired, we reject the connection
console.error("Rejected invalid or expired token:", err.message);
data.connectionConfig.isAuthenticated = false;
throw new Error("Token invalid or expired");
}
console.log(`Checking token for constellation ${constellation_uuid} for user ${token}`);
await axiosClient.get(`${POSTGRES_HOSTNAME}/me/constellations/${constellation_uuid}/access`, {
headers: {
"Authorization": `Bearer ${token}`
}
})
.then(response => {
const permission = DATA_PERMISSION.indexOf(response.data.access); // Get the permission level from the response
if (permission >= DATA_PERMISSION.indexOf("WRITE")) {
data.connectionConfig.isAuthenticated = true; // Set authenticated flag
data.connectionConfig.readOnly = false; // Set read-only flag to false
data.connectionConfig.token = token; // Store the token for write access
} else if (permission >= DATA_PERMISSION.indexOf("READ")) {
data.connectionConfig.isAuthenticated = true; // Set authenticated flag for read-only access
data.connectionConfig.readOnly = true; // Set read-only flag
data.connectionConfig.token = token; // Store the token for read-only access
} else E{
console.log(`Connection from ${socketId} rejected due to insufficient permissions`);
data.connectionConfig.isAuthenticated = false; // Set authenticated flag to false
throw new Error("Insufficient permissions");
}
}
)
.catch(error => {
console.error(`Error validating token:`, error);
data.connectionConfig.isAuthenticated = false; // Set authenticated flag to false
throw new Error("Token validation failed");
});
return {
"user": {
"token": token,
"sub": decodedToken.sub, // User UUID from the token
"exp": decodedToken.exp, // Expiration time from the token
}
}
},
// Someone is connected to the server
async connected(data) {
console.log("connected");
const { connectionConfig } = data;
if (connectionConfig.isAuthenticated) {
console.log("A new authenticated connection has been established. connections:", server.hocuspocus.getConnectionsCount());
} else E{
throw new Error("Connection not authenticated"); // Reject the connection if not authenticated
}
},
// When a message is send to the server (e.g. a CRDT operation) (useful to check if the token is expired)
async beforeHandleMessage(data) {
console.log("beforeHandleMessage");
try {
const currentTime = Math.floor(Date.now() / 1000); // Current time in seconds
// Check if the token has expired
Iif (data.context.user.exp < currentTime) {
console.log("Token has expired, rejecting message.");
throw new Error("Token has expired");
}
} catch (error) {
console.error("Error decoding token:", error);
throw new Error("Invalid token");
}
},
// When the document is being loaded (should ask the api for the document content)
async onLoadDocument(data) {
console.log("onDocumentLoad");
const { documentName } = data;
console.log(`Document ${documentName} is being loaded.`);
const documentNameList = documentName.split("/");
const apiType = documentNameList[0]; // Get the API type from the document name (postgres or neo4j)
const constellation_uuid = documentNameList[1]; // Get the constellation UUID from the document name
let collection_uuid;
let document_uuid;
let document_attribute;
let neo4j_object_type;
if (apiType === "postgres") {
collection_uuid = documentNameList[2]; // Get the collection UUID from the document name
document_uuid = documentNameList[3]; // Get the document UUID from the document name
} else if (apiType.startsWith("neo4j")) {
document_uuid = documentNameList[2]; // Get the document UUID from the document name
document_attribute = documentNameList[3]; // Get the document attribute from the document name
if (apiType === "neo4j_node") {
neo4j_object_type = "node"; // Set the Neo4j object type to node
} else if (apiType === "neo4j_link") {
neo4j_object_type = "link"; // Set the Neo4j object type to link
} else E{
throw new Error("Invalid API type");
}
} else E{
throw new Error("Invalid API type");
}
if (apiType === "postgres") {
// Load the document from the Postgres API
await axiosClient.get(`${POSTGRES_HOSTNAME}/collections/${collection_uuid}/ydocs/${document_uuid}/content`, {
headers: {
"Authorization": `Bearer ${data.connectionConfig.token}`
}
})
.then(response => {
const content = response.data.content; // Assuming the content is in the response data
if (!content || content.length === 0) {
// If the content is empty, return a new Yjs document
return new Y.Doc(); // Return an empty Yjs document
}
// Decode from Base64 back to a Uint8Array
const update = Buffer.from(content, "base64");
// Apply it to the Y.Doc
Y.applyUpdate(data.document, update);
})
.catch(error => {
console.error(`Error loading document from Postgres API:`, error);
throw new Error("Failed to load document from Postgres API");
});
} else if (apiType.startsWith("neo4j")) {
// Load the document from the Neo4j API
await axiosClient.get(`${NEO4J_HOSTNAME}/constellation/${constellation_uuid}/${neo4j_object_type}/${document_uuid}/attribute/${document_attribute}`, {
headers: {
"Authorization": `Bearer ${data.connectionConfig.token}`
}
})
.then(response => {
const content = response.data.data[0]; // Assuming the content is in the response data
if (!content || content.length === 0 || !content.startsWith("ydoc:")) {
// If the content is empty, return a new Yjs document
return new Y.Doc(); // Return an empty Yjs document
}
// Decode from Base64 back to a Uint8Array
const update = Buffer.from(content.slice(5), "base64"); // Remove the "ydoc:" prefix
// Apply it to the Y.Doc
Y.applyUpdate(data.document, update);
})
.catch(error => {
console.error(`Error loading document from Neo4j API:`, error);
throw new Error("Failed to load document from Neo4j API");
});
} else E{
throw new Error("Invalid API type");
}
},
// If we need to do something after the document is loaded
async afterLoadDocument(data) {
console.log("afterLoadDocument");
const { documentName } = data;
console.log(`Document ${documentName} has been loaded.`);
},
// When a document gets some changes (not debounced so WARNING: this can be called very often: multiple times per second)
async onChange(data) {
console.log("onChange");
const { documentName } = data;
console.log(`Document ${documentName} has been changed.`);
},
// When a document gets some changes (debounced so WARNING: this is called only after "debounce" time of inactivity or when "maxDebounce" is reached)
async onStoreDocument(data) {
console.log("onStoreDocument");
const { documentName } = data;
console.log(`Document ${documentName} is being stored.`);
const documentNameList = documentName.split("/");
const apiType = documentNameList[0]; // Get the API type from the document name (postgres or neo4j)
const constellation_uuid = documentNameList[1]; // Get the constellation UUID from the document name
let collection_uuid;
let document_uuid;
let document_attribute;
let neo4j_object_type;
if (apiType === "postgres") {
collection_uuid = documentNameList[2]; // Get the collection UUID from the document name
document_uuid = documentNameList[3]; // Get the document UUID from the document name
} else if (apiType.startsWith("neo4j")) {
document_uuid = documentNameList[2]; // Get the document UUID from the document name
document_attribute = documentNameList[3]; // Get the document attribute from the document name
if (apiType === "neo4j_node") {
neo4j_object_type = "node"; // Set the Neo4j object type to node
} else if (apiType === "neo4j_link") {
neo4j_object_type = "link"; // Set the Neo4j object type to link
} else E{
throw new Error("Invalid API type");
}
} else E{
throw new Error("Invalid API type");
}
if (apiType === "postgres") {
// 1. encode the full Y.Doc state (including history) as a Uint8Array
const update = Y.encodeStateAsUpdate(data.document);
// 2. convert to Base64 so it’s safe to send as JSON
const payload = Buffer.from(update).toString("base64");
const json = {
"content": payload,
};
await axiosClient.patch(`${POSTGRES_HOSTNAME}/collections/${collection_uuid}/ydocs/${document_uuid}/content`,
json,
{
headers: {
"Authorization": `Bearer ${data.context.user.token}`
}
}
)
.then(response => {
console.log(`Document ${document_uuid} has been updated in Postgres API.`);
})
.catch(error => {
console.error(`Error updating document in Postgres API:`, error);
throw new Error("Failed to update document in Postgres API");
});
} else if (apiType.startsWith("neo4j")) {
// 1. encode the full Y.Doc state (including history) as a Uint8Array
const update = Y.encodeStateAsUpdate(data.document);
// 2. convert to Base64 so it’s safe to send as JSON
const payload = Buffer.from(update).toString("base64");
const json = {
"value": "ydoc:" + payload,
}
await axiosClient.patch(`${NEO4J_HOSTNAME}/constellation/${constellation_uuid}/${neo4j_object_type}/${document_uuid}/attribute/${document_attribute}`,
json,
{
headers: {
"Authorization": `Bearer ${data.context.user.token}`
}
}
)
.then(response => {
console.log(`Document ${document_uuid} has been updated in Neo4j API.`);
})
.catch(error => {
console.error(`Error updating document in Neo4j API:`, error);
throw new Error("Failed to update document in Neo4j API");
});
} else E{
throw new Error("Invalid API type");
}
},
});
// // … and run it!
// server.listen();
|