fix: live restructuring
This commit is contained in:
Palanikannan M
@@ -0,0 +1,17 @@
|
||||
// Export all controllers from this barrel file
|
||||
import { CollaborationController } from "./collaboration.controller";
|
||||
import { LiveDocumentController } from "./live-document.controller";
|
||||
import { DocumentController } from "./document.controller";
|
||||
import { HealthController } from "./health.controller";
|
||||
|
||||
export const CONTROLLERS = {
|
||||
// Core system controllers (health checks, status endpoints)
|
||||
CORE: [HealthController],
|
||||
|
||||
// Document management controllers
|
||||
DOCUMENT: [DocumentController, LiveDocumentController],
|
||||
|
||||
// WebSocket controllers for real-time functionality
|
||||
WEBSOCKET: [CollaborationController],
|
||||
};
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
import { Controller, Get } from "@plane/decorators";
|
||||
import type { Request, Response } from "express";
|
||||
import { z } from "zod";
|
||||
import * as Y from "yjs";
|
||||
|
||||
// Server agent
|
||||
import { serverAgentManager } from "@/core/agents/server-agent";
|
||||
|
||||
// Helpers
|
||||
import { handleError } from "@/core/helpers/error-handling/error-factory";
|
||||
import { AppError } from "@/core/helpers/error-handling/error-handler";
|
||||
import { env } from "@/env";
|
||||
|
||||
// Types
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
import { manualLogger } from "@/core/helpers/logger";
|
||||
import { getAllDocumentFormatsFromDocumentEditorBinaryData } from "@plane/editor";
|
||||
|
||||
// Schema for request validation
|
||||
const getLiveDocumentValuesSchema = z.object({
|
||||
documentId: z.string().min(1, "Document ID is required"),
|
||||
variant: z.enum(["document"]),
|
||||
workspaceSlug: z.string().min(1, "Workspace slug is required"),
|
||||
});
|
||||
|
||||
@Controller("/live-document")
|
||||
export class LiveDocumentController {
|
||||
@Get("/")
|
||||
async getLiveDocumentValues(req: Request, res: Response) {
|
||||
const requestId = req.id || crypto.randomUUID();
|
||||
|
||||
try {
|
||||
if (req.headers["live-server-secret-key"] !== env.LIVE_SERVER_SECRET_KEY) {
|
||||
return res.status(401).json({
|
||||
message: "Unauthorized access",
|
||||
status: 401,
|
||||
context: {
|
||||
component: "get-live-document-values-controller",
|
||||
operation: "getLiveDocumentValues",
|
||||
requestId,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const validatedData = getLiveDocumentValuesSchema.parse(req.query);
|
||||
const { documentId, workspaceSlug } = validatedData;
|
||||
|
||||
const context: Partial<HocusPocusServerContext> = {
|
||||
workspaceSlug,
|
||||
};
|
||||
|
||||
try {
|
||||
const { connection } = await serverAgentManager.getConnection(documentId, context);
|
||||
|
||||
// Define the document type
|
||||
type DocumentData = {
|
||||
description_binary: string;
|
||||
description: object;
|
||||
description_html: string;
|
||||
name: string;
|
||||
};
|
||||
|
||||
// Create a promise to wrap the setTimeout
|
||||
const loadDocumentWithDelay = new Promise<DocumentData | null>((resolve) => {
|
||||
let documentData: DocumentData;
|
||||
|
||||
connection.transact((doc) => {
|
||||
const type = doc.getXmlFragment("default");
|
||||
const contentDoc = type.doc;
|
||||
|
||||
if (!contentDoc) {
|
||||
resolve(null);
|
||||
return;
|
||||
}
|
||||
|
||||
const yjsBinary = Y.encodeStateAsUpdate(contentDoc);
|
||||
const { contentBinaryEncoded, contentJSON, contentHTML, titleHTML } =
|
||||
getAllDocumentFormatsFromDocumentEditorBinaryData(yjsBinary);
|
||||
|
||||
documentData = {
|
||||
description_binary: contentBinaryEncoded,
|
||||
description: contentJSON,
|
||||
description_html: contentHTML,
|
||||
name: titleHTML,
|
||||
};
|
||||
|
||||
resolve(documentData);
|
||||
});
|
||||
});
|
||||
|
||||
// Await the delayed document loading
|
||||
const documentLoaded = await loadDocumentWithDelay;
|
||||
|
||||
await serverAgentManager.releaseConnection(documentId);
|
||||
|
||||
// Return the converted document
|
||||
res.status(200).json(documentLoaded);
|
||||
} catch (error) {
|
||||
// Error during server agent connection or conversion
|
||||
manualLogger.error(`Error processing document ${documentId}:`, error);
|
||||
|
||||
res.status(400).json({
|
||||
loaded: false,
|
||||
message: "Document not currently loaded in memory",
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
let appError: AppError;
|
||||
|
||||
if (error instanceof z.ZodError) {
|
||||
// Handle validation errors
|
||||
appError = handleError(error, {
|
||||
errorType: "unprocessable-entity",
|
||||
message: "Invalid request data",
|
||||
component: "get-live-document-values-controller",
|
||||
operation: "getLiveDocumentValues",
|
||||
extraContext: {
|
||||
requestId,
|
||||
validationErrors: error.errors.map((err) => ({
|
||||
path: err.path.join("."),
|
||||
message: err.message,
|
||||
})),
|
||||
},
|
||||
});
|
||||
} else {
|
||||
// Handle other errors
|
||||
appError = handleError(error, {
|
||||
errorType: "internal",
|
||||
message: "Internal server error",
|
||||
component: "get-live-document-values-controller",
|
||||
operation: "getLiveDocumentValues",
|
||||
extraContext: { requestId },
|
||||
});
|
||||
}
|
||||
|
||||
res.status(appError.status).json({
|
||||
message: appError.message,
|
||||
status: appError.status,
|
||||
context: appError.context,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
import * as Y from "yjs";
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
|
||||
export class DocumentProcessor {
|
||||
static async process(
|
||||
xmlFragment: Y.XmlFragment,
|
||||
pageId: string,
|
||||
context: HocusPocusServerContext,
|
||||
options: {
|
||||
targetNodeId?: string;
|
||||
componentType?: string;
|
||||
[key: string]: any;
|
||||
} = {}
|
||||
): Promise<void> {}
|
||||
}
|
||||
@@ -0,0 +1,425 @@
|
||||
import * as Y from "yjs";
|
||||
import { Response } from "express";
|
||||
|
||||
import { DirectConnection, Hocuspocus } from "@hocuspocus/server";
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import { manualLogger } from "@/core/helpers/logger";
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
import { DocumentProcessor } from "@/plane-live/lib/document-processor";
|
||||
|
||||
/**
|
||||
* Metadata for a stored connection
|
||||
*/
|
||||
interface ConnectionData {
|
||||
connection: DirectConnection;
|
||||
context: Partial<HocusPocusServerContext>;
|
||||
createdAt: number;
|
||||
lastUsed: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Connection statistics for a document
|
||||
*/
|
||||
interface ConnectionStats {
|
||||
documentId: string;
|
||||
createdAt: string;
|
||||
lastUsed: string;
|
||||
idleTime: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Overall statistics for the connection manager
|
||||
*/
|
||||
interface ManagerStats {
|
||||
totalConnections: number;
|
||||
connections: ConnectionStats[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Error thrown when the ServerAgentManager is not properly initialized
|
||||
*/
|
||||
class ServerAgentManagerError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "ServerAgentManagerError";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Manages server-side connections (agents) to the Hocuspocus server
|
||||
* Implements the Singleton pattern to ensure only one instance exists
|
||||
*/
|
||||
export class ServerAgentManager {
|
||||
private static instance: ServerAgentManager;
|
||||
private connections: Map<string, ConnectionData>;
|
||||
public hocuspocusServer: Hocuspocus | null;
|
||||
private cleanupInterval: ReturnType<typeof setTimeout> | null;
|
||||
|
||||
/**
|
||||
* Private constructor to enforce singleton pattern
|
||||
*/
|
||||
private constructor() {
|
||||
this.connections = new Map<string, ConnectionData>();
|
||||
this.hocuspocusServer = null;
|
||||
this.cleanupInterval = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the singleton instance of the ServerAgentManager
|
||||
* @returns {ServerAgentManager} The singleton instance
|
||||
*/
|
||||
public static getInstance(): ServerAgentManager {
|
||||
if (!ServerAgentManager.instance) {
|
||||
ServerAgentManager.instance = new ServerAgentManager();
|
||||
}
|
||||
return ServerAgentManager.instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the manager with a Hocuspocus server instance
|
||||
* @param {Hocuspocus} server - The Hocuspocus server instance
|
||||
* @returns {ServerAgentManager} The initialized manager instance for chaining
|
||||
*/
|
||||
public initialize(server: Hocuspocus): ServerAgentManager {
|
||||
this.hocuspocusServer = server;
|
||||
|
||||
// Set up periodic cleanup of unused connections
|
||||
this.startConnectionCleanup();
|
||||
|
||||
manualLogger.info("ServerAgentManager initialized");
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get or create a connection to a document
|
||||
* @param {string} documentId - The document ID to connect to
|
||||
* @param {HocusPocusServerContext} context - Additional context for the connection
|
||||
* @returns {Promise<ConnectionData>} - The connection data object
|
||||
* @throws {ServerAgentManagerError} If the manager is not initialized
|
||||
*/
|
||||
public async getConnection(documentId: string, context: Partial<HocusPocusServerContext>): Promise<ConnectionData> {
|
||||
if (!this.hocuspocusServer) {
|
||||
throw new ServerAgentManagerError("ServerAgentManager not initialized with a Hocuspocus server");
|
||||
}
|
||||
|
||||
// Check if we already have a connection for this document
|
||||
if (this.connections.has(documentId)) {
|
||||
// manualLogger.info(`Reusing existing connection for document: ${documentId}`);
|
||||
const connectionData = this.connections.get(documentId)!;
|
||||
// Update last used timestamp
|
||||
connectionData.lastUsed = Date.now();
|
||||
return connectionData;
|
||||
}
|
||||
|
||||
context.documentType = context.documentType === "sync_agent" ? "sync_agent" : "server_agent";
|
||||
try {
|
||||
// Create a new connection
|
||||
const connection = await this.hocuspocusServer.openDirectConnection(documentId, {
|
||||
documentType: context.documentType,
|
||||
projectId: context.projectId,
|
||||
workspaceSlug: context.workspaceSlug,
|
||||
// triggerExecutionAfterLoad: context.triggerExecutionAfterLoad,
|
||||
agentId: uuidv4(), // Unique ID for this server agent
|
||||
});
|
||||
|
||||
// Store the connection with metadata
|
||||
const connectionData: ConnectionData = {
|
||||
connection,
|
||||
context,
|
||||
createdAt: Date.now(),
|
||||
lastUsed: Date.now(),
|
||||
};
|
||||
|
||||
this.connections.set(documentId, connectionData);
|
||||
|
||||
return connectionData;
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
manualLogger.error(`Failed to create connection for document ${documentId}: ${errorMessage}`);
|
||||
throw new ServerAgentManagerError(`Failed to create connection: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a transaction on a document
|
||||
* @param {string} documentId - The document ID
|
||||
* @param {(doc: Y.Doc) => void | Promise<void>} transactionFn - The transaction function
|
||||
* @param {HocusPocusServerContext} context - Additional context for the connection
|
||||
* @returns {Promise<void>} - Promise that resolves when the transaction is complete
|
||||
* @throws {ServerAgentManagerError} If the transaction fails
|
||||
*/
|
||||
public async executeTransaction(
|
||||
documentId: string,
|
||||
transactionFn: (_doc: Y.Doc) => void | Promise<void>,
|
||||
context: Partial<HocusPocusServerContext>,
|
||||
res?: Response
|
||||
): Promise<boolean> {
|
||||
let connectionData: ConnectionData | null = null;
|
||||
|
||||
try {
|
||||
connectionData = await this.getConnection(documentId, context);
|
||||
connectionData.lastUsed = Date.now();
|
||||
|
||||
connectionData.context = { ...connectionData.context };
|
||||
// Execute the transaction
|
||||
await connectionData.connection.transact(transactionFn);
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
manualLogger.error(`Transaction error for document ${documentId}: ${errorMessage}`, {
|
||||
documentId,
|
||||
error: error instanceof Error ? error.stack : errorMessage,
|
||||
context,
|
||||
});
|
||||
|
||||
// Notify about transaction failure if needed
|
||||
this.notifyTransactionFailure(documentId, errorMessage, res);
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public notifySyncTrigger(
|
||||
pageId: string,
|
||||
context: HocusPocusServerContext,
|
||||
options: {
|
||||
componentType?: string;
|
||||
targetNodeId?: string;
|
||||
[key: string]: any;
|
||||
} = {}
|
||||
) {
|
||||
if (!this.hocuspocusServer) return;
|
||||
|
||||
this.executeTransaction(
|
||||
pageId,
|
||||
async (doc) => {
|
||||
const xmlFragment = doc.getXmlFragment("default");
|
||||
|
||||
// Process the document using our extensible system
|
||||
DocumentProcessor.process(xmlFragment, pageId, context, options);
|
||||
},
|
||||
{
|
||||
...context,
|
||||
documentType: "sync_agent",
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Notify clients about a successful transaction
|
||||
* @private
|
||||
* @param {string} documentId - The document ID
|
||||
*/
|
||||
public notifyTransactionSuccess(documentId: string, res?: Response): void {
|
||||
if (!this.hocuspocusServer) return;
|
||||
|
||||
const document = this.hocuspocusServer.documents.get(documentId);
|
||||
if (!document) return;
|
||||
|
||||
try {
|
||||
manualLogger.info(`notified transaction success for ${documentId}:`);
|
||||
// res.status(200).json({ success: true });
|
||||
} catch (error) {
|
||||
manualLogger.error(`Error notifying transaction success for ${documentId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Notify clients about a failed transaction
|
||||
* @private
|
||||
* @param {string} documentId - The document ID
|
||||
* @param {string} errorMessage - The error message
|
||||
*/
|
||||
private notifyTransactionFailure(documentId: string, errorMessage: string, res?: Response): void {
|
||||
if (!this.hocuspocusServer) return;
|
||||
|
||||
const document = this.hocuspocusServer.documents.get(documentId);
|
||||
if (!document) return;
|
||||
|
||||
try {
|
||||
// res.status(200).json({ success: true });
|
||||
// document.broadcastStateless(
|
||||
// JSON.stringify({
|
||||
// type: "transaction_status",
|
||||
// status: "error",
|
||||
// message: errorMessage,
|
||||
// timestamp: new Date().toISOString(),
|
||||
// })
|
||||
// );
|
||||
} catch (error) {
|
||||
manualLogger.error(`Error notifying transaction failure for ${documentId}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Release a connection when it's no longer needed
|
||||
* @param {string} documentId - The document ID
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
public async releaseConnection(documentId: string): Promise<void> {
|
||||
if (!this.connections.has(documentId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const connectionData = this.connections.get(documentId)!;
|
||||
|
||||
try {
|
||||
connectionData.connection.disconnect();
|
||||
this.connections.delete(documentId);
|
||||
manualLogger.info(`Released connection for document: ${documentId}`);
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
manualLogger.error(`Error releasing connection for document ${documentId}: ${errorMessage}`);
|
||||
// We still want to remove it from our map even if disconnect fails
|
||||
this.connections.delete(documentId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a document has any client connections (excluding our agent)
|
||||
* @param {string} documentId - The document ID
|
||||
* @returns {boolean} - True if there are client connections
|
||||
*/
|
||||
/**
|
||||
* Check if a document has any client connections (excluding our agent)
|
||||
* @param {string} documentId - The document ID
|
||||
* @returns {boolean} - True if there are client connections
|
||||
*/
|
||||
public hasClientConnections(documentId: string): boolean {
|
||||
if (!this.hocuspocusServer) return false;
|
||||
|
||||
// Get the document from the server
|
||||
const document = this.hocuspocusServer.documents.get(documentId);
|
||||
if (!document) return false;
|
||||
return document.connections.size > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Start periodic cleanup of unused connections
|
||||
* @private
|
||||
*/
|
||||
private startConnectionCleanup(): void {
|
||||
// Check every 5 minutes for unused connections
|
||||
const CLEANUP_INTERVAL = 5 * 60 * 1000;
|
||||
|
||||
// Clear any existing interval
|
||||
if (this.cleanupInterval) {
|
||||
clearInterval(this.cleanupInterval);
|
||||
}
|
||||
|
||||
this.cleanupInterval = setInterval(() => {
|
||||
this.cleanupUnusedConnections().catch((error) => {
|
||||
manualLogger.error("Error during connection cleanup:", error);
|
||||
});
|
||||
}, CLEANUP_INTERVAL);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up connections that are no longer needed
|
||||
* @private
|
||||
*/
|
||||
private async cleanupUnusedConnections(): Promise<void> {
|
||||
const documentsToCleanup: string[] = [];
|
||||
|
||||
for (const [documentId] of this.connections.entries()) {
|
||||
// If no client connections exist (only our agent remains), schedule cleanup
|
||||
if (!this.hasClientConnections(documentId)) {
|
||||
documentsToCleanup.push(documentId);
|
||||
}
|
||||
}
|
||||
|
||||
// Release connections outside the loop to avoid modifying the map during iteration
|
||||
for (const documentId of documentsToCleanup) {
|
||||
manualLogger.info(`Cleaning up connection for document with no clients: ${documentId}`);
|
||||
await this.releaseConnection(documentId);
|
||||
}
|
||||
|
||||
if (documentsToCleanup.length > 0) {
|
||||
manualLogger.info(`Cleaned up ${documentsToCleanup.length} connections with no clients`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a document has no client connections and release the agent if needed
|
||||
* @param {string} documentId - The document ID to check
|
||||
* @returns {Promise<boolean>} - True if the connection was released
|
||||
*/
|
||||
public async checkAndReleaseIfNoClients(documentId: string): Promise<boolean> {
|
||||
if (!this.connections.has(documentId)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!this.hasClientConnections(documentId)) {
|
||||
manualLogger.info(`No clients left for document ${documentId}, releasing agent connection`);
|
||||
await this.releaseConnection(documentId);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up hooks on the HocusPocus server to trigger cleanup when clients disconnect
|
||||
* @returns {ServerAgentManager} The manager instance for chaining
|
||||
*/
|
||||
public setupHocusPocusHooks(): ServerAgentManager {
|
||||
if (!this.hocuspocusServer) {
|
||||
throw new ServerAgentManagerError("ServerAgentManager not initialized with a Hocuspocus server");
|
||||
}
|
||||
|
||||
// Add a hook to the onDisconnect event
|
||||
this.hocuspocusServer.configure({
|
||||
async onDisconnect({ documentName }) {
|
||||
// Use a small delay to ensure the connection is fully closed before checking
|
||||
setTimeout(async () => {
|
||||
try {
|
||||
await serverAgentManager.checkAndReleaseIfNoClients(documentName);
|
||||
} catch (error) {
|
||||
manualLogger.error(`Error checking client connections for ${documentName}:`, error);
|
||||
}
|
||||
}, 100); // Small delay to ensure connection state is updated
|
||||
},
|
||||
});
|
||||
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* Get statistics about current connections
|
||||
* @returns {ManagerStats} - Connection statistics
|
||||
*/
|
||||
public getStats(): ManagerStats {
|
||||
return {
|
||||
totalConnections: this.connections.size,
|
||||
connections: Array.from(this.connections.entries()).map(([documentId, data]) => ({
|
||||
documentId,
|
||||
createdAt: new Date(data.createdAt).toISOString(),
|
||||
lastUsed: new Date(data.lastUsed).toISOString(),
|
||||
idleTime: Math.round((Date.now() - data.lastUsed) / 1000) + "s",
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Shutdown the manager and clean up all connections
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
public async shutdown(): Promise<void> {
|
||||
// Clear the cleanup interval
|
||||
if (this.cleanupInterval) {
|
||||
clearInterval(this.cleanupInterval);
|
||||
this.cleanupInterval = null;
|
||||
}
|
||||
|
||||
// Release all connections
|
||||
const documentIds = Array.from(this.connections.keys());
|
||||
for (const documentId of documentIds) {
|
||||
await this.releaseConnection(documentId);
|
||||
}
|
||||
|
||||
manualLogger.info("ServerAgentManager shut down successfully");
|
||||
}
|
||||
}
|
||||
|
||||
// Create and export the singleton instance
|
||||
export const serverAgentManager = ServerAgentManager.getInstance();
|
||||
@@ -1,21 +1,4 @@
|
||||
import { HealthController } from "@/core/controllers/health.controller";
|
||||
import { DocumentController } from "@/core/controllers/document.controller";
|
||||
import { CollaborationController } from "@/core/controllers/collaboration.controller";
|
||||
|
||||
/**
|
||||
* Controller registry exports
|
||||
* Simple grouped arrays of controller classes for better organization
|
||||
*/
|
||||
export const CONTROLLERS = {
|
||||
// Core system controllers (health checks, status endpoints)
|
||||
CORE: [HealthController],
|
||||
|
||||
// Document management controllers
|
||||
DOCUMENT: [DocumentController],
|
||||
|
||||
// WebSocket controllers for real-time functionality
|
||||
WEBSOCKET: [CollaborationController],
|
||||
};
|
||||
import { CONTROLLERS } from "@/plane-live/controllers";
|
||||
|
||||
// Helper to get all REST controllers
|
||||
export const getAllControllers = () => [...CONTROLLERS.CORE, ...CONTROLLERS.DOCUMENT, ...CONTROLLERS.WEBSOCKET];
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
// Export all controllers from this barrel file
|
||||
export { HealthController } from "./health.controller";
|
||||
export { CollaborationController } from "./collaboration.controller";
|
||||
export { DocumentController } from "./document.controller";
|
||||
@@ -0,0 +1,233 @@
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
import { BasePageService } from "@/core/services/base-page.service";
|
||||
import { logger } from "@plane/logger";
|
||||
import { getAllDocumentFormatsFromBinaryData, getBinaryDataFromHTMLString } from "@/core/helpers/page";
|
||||
import { TPage } from "@plane/types";
|
||||
import { DocumentHandler, HandlerDefinition } from "@/core/types/document-handler";
|
||||
import { handlerFactory } from "@/core/handlers/page-handlers/handler-factory";
|
||||
|
||||
/**
|
||||
* Base class for page handlers with factory integration
|
||||
*/
|
||||
export abstract class BasePageHandler<TService extends BasePageService, TConfig extends Record<string, any>> {
|
||||
/**
|
||||
* The document type identifier
|
||||
*/
|
||||
protected abstract documentType: string;
|
||||
|
||||
constructor(protected service: TService) {}
|
||||
|
||||
/**
|
||||
* Abstract method to get config from context
|
||||
*/
|
||||
protected abstract getConfig(context: HocusPocusServerContext): TConfig;
|
||||
|
||||
/**
|
||||
* Fetches the binary description data for a page
|
||||
*/
|
||||
public async fetchPageDescriptionBinary({
|
||||
pageId,
|
||||
context,
|
||||
}: {
|
||||
pageId: string;
|
||||
context: HocusPocusServerContext;
|
||||
}): Promise<Uint8Array | undefined> {
|
||||
const { cookie } = context;
|
||||
const config = this.getConfig(context);
|
||||
|
||||
if (!pageId) return;
|
||||
|
||||
const response = await this.service.fetchDescriptionBinary({
|
||||
pageId,
|
||||
cookie,
|
||||
config,
|
||||
});
|
||||
const binaryData = new Uint8Array(response);
|
||||
|
||||
if (binaryData.byteLength === 0) {
|
||||
const binary = await this.transformHTMLToBinary(config, pageId, cookie);
|
||||
if (binary) {
|
||||
return binary;
|
||||
}
|
||||
}
|
||||
|
||||
return binaryData;
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates the description of a page
|
||||
*/
|
||||
public async updatePageDescription({
|
||||
pageId,
|
||||
state: updatedDescription,
|
||||
title,
|
||||
context,
|
||||
}: {
|
||||
pageId: string;
|
||||
state: Uint8Array;
|
||||
title: string;
|
||||
context: HocusPocusServerContext;
|
||||
}): Promise<void> {
|
||||
if (!(updatedDescription instanceof Uint8Array)) {
|
||||
throw new Error("Invalid updatedDescription: must be an instance of Uint8Array");
|
||||
}
|
||||
|
||||
const { cookie } = context;
|
||||
const config = this.getConfig(context);
|
||||
|
||||
if (!pageId) return;
|
||||
|
||||
const { contentBinaryEncoded, contentHTML, contentJSON } = getAllDocumentFormatsFromBinaryData(updatedDescription);
|
||||
const payload = {
|
||||
description_binary: contentBinaryEncoded,
|
||||
description_html: contentHTML,
|
||||
description: contentJSON,
|
||||
name: title,
|
||||
};
|
||||
|
||||
await this.service.updateDescription({
|
||||
config,
|
||||
pageId,
|
||||
data: payload,
|
||||
cookie,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches the title of a page
|
||||
*/
|
||||
public async fetchPageTitle({
|
||||
context,
|
||||
pageId,
|
||||
}: {
|
||||
pageId: string;
|
||||
context: HocusPocusServerContext;
|
||||
}): Promise<string | undefined> {
|
||||
const { cookie } = context;
|
||||
const config = this.getConfig(context);
|
||||
|
||||
if (!pageId) return;
|
||||
|
||||
try {
|
||||
const pageDetails = await this.service.fetchDetails({
|
||||
config,
|
||||
pageId,
|
||||
cookie,
|
||||
});
|
||||
return pageDetails.name;
|
||||
} catch (error) {
|
||||
logger.error("Error while fetching page title", error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates the title of a page
|
||||
*/
|
||||
public async updatePageTitle({
|
||||
context,
|
||||
pageId,
|
||||
title,
|
||||
abortSignal,
|
||||
}: {
|
||||
pageId: string;
|
||||
title: string;
|
||||
abortSignal?: AbortSignal;
|
||||
context: HocusPocusServerContext;
|
||||
}): Promise<void> {
|
||||
const { cookie } = context;
|
||||
const config = this.getConfig(context);
|
||||
|
||||
if (!pageId) return;
|
||||
|
||||
const payload = {
|
||||
name: title,
|
||||
};
|
||||
|
||||
await this.service.updateTitle({
|
||||
config,
|
||||
pageId,
|
||||
data: payload,
|
||||
cookie,
|
||||
abortSignal,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches sub-page details
|
||||
*/
|
||||
public async fetchPageSubPageDetails({
|
||||
context,
|
||||
pageId,
|
||||
}: {
|
||||
pageId: string;
|
||||
context: HocusPocusServerContext;
|
||||
}): Promise<TPage[] | undefined> {
|
||||
const { cookie } = context;
|
||||
const config = this.getConfig(context);
|
||||
|
||||
if (!pageId) return;
|
||||
|
||||
try {
|
||||
const response = await this.service.fetchSubPageDetails({
|
||||
config,
|
||||
pageId,
|
||||
cookie,
|
||||
});
|
||||
return response;
|
||||
} catch (error) {
|
||||
logger.error("Fetch error:", error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Transforms HTML to binary data
|
||||
*/
|
||||
private async transformHTMLToBinary(
|
||||
config: TConfig,
|
||||
pageId: string,
|
||||
cookie: string
|
||||
): Promise<Uint8Array | undefined> {
|
||||
try {
|
||||
const pageDetails = await this.service.fetchDetails({
|
||||
config,
|
||||
pageId,
|
||||
cookie,
|
||||
});
|
||||
|
||||
const { contentBinary } = getBinaryDataFromHTMLString(pageDetails.description_html ?? "<p></p>");
|
||||
return contentBinary;
|
||||
} catch (error) {
|
||||
logger.error("Error while transforming from HTML to Uint8Array", error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the handler definition
|
||||
*/
|
||||
protected createHandlerDefinition(): HandlerDefinition {
|
||||
const handler: DocumentHandler = {
|
||||
fetch: this.fetchPageDescriptionBinary.bind(this),
|
||||
store: this.updatePageDescription.bind(this),
|
||||
fetchTitle: this.fetchPageTitle.bind(this),
|
||||
updateTitle: this.updatePageTitle.bind(this),
|
||||
fetchSubPages: this.fetchPageSubPageDetails.bind(this),
|
||||
};
|
||||
|
||||
return {
|
||||
selector: (context: Partial<HocusPocusServerContext>) => context.documentType === this.documentType,
|
||||
handler,
|
||||
priority: 10, // Standard priority
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers the handler with the handler factory
|
||||
*/
|
||||
public register(): void {
|
||||
const definition = this.createHandlerDefinition();
|
||||
handlerFactory.register(definition);
|
||||
}
|
||||
}
|
||||
Binary file not shown.
@@ -1,27 +1,28 @@
|
||||
import { Database } from "@hocuspocus/extension-database";
|
||||
import { catchAsync } from "@/core/helpers/error-handling/error-handler";
|
||||
import { handleError } from "@/core/helpers/error-handling/error-factory";
|
||||
import { getDocumentHandler } from "../handlers/document-handlers";
|
||||
import { type HocusPocusServerContext, type TDocumentTypes } from "@/core/types/common";
|
||||
import { storePayload } from "@hocuspocus/server";
|
||||
import { extractTextFromHTML } from "./title-update/title-utils";
|
||||
import { getDocumentHandler } from "../handlers/page-handlers";
|
||||
|
||||
export const createDatabaseExtension = () => {
|
||||
return new Database({
|
||||
fetch: handleFetch,
|
||||
store: handleStore,
|
||||
store: handleStore as (data: storePayload) => Promise<void>,
|
||||
});
|
||||
};
|
||||
|
||||
const handleFetch = async ({
|
||||
context,
|
||||
documentName: pageId,
|
||||
requestParameters,
|
||||
documentName,
|
||||
}: {
|
||||
context: HocusPocusServerContext;
|
||||
documentName: TDocumentTypes;
|
||||
documentName: string;
|
||||
requestParameters: URLSearchParams;
|
||||
}) => {
|
||||
const { documentType } = context;
|
||||
const params = requestParameters;
|
||||
const pageId = documentName as TDocumentTypes;
|
||||
|
||||
let fetchedData = null;
|
||||
fetchedData = await catchAsync(
|
||||
@@ -41,7 +42,6 @@ const handleFetch = async ({
|
||||
fetchedData = await documentHandler.fetch({
|
||||
context: context as HocusPocusServerContext,
|
||||
pageId,
|
||||
params,
|
||||
});
|
||||
|
||||
if (!fetchedData) {
|
||||
@@ -67,15 +67,20 @@ const handleFetch = async ({
|
||||
const handleStore = async ({
|
||||
context,
|
||||
state,
|
||||
documentName: pageId,
|
||||
requestParameters,
|
||||
}: {
|
||||
documentName,
|
||||
document,
|
||||
}: Partial<storePayload> & {
|
||||
context: HocusPocusServerContext;
|
||||
state: Buffer;
|
||||
documentName: TDocumentTypes;
|
||||
requestParameters: URLSearchParams;
|
||||
documentName: string;
|
||||
}) => {
|
||||
catchAsync(
|
||||
const pageId = documentName;
|
||||
|
||||
if (!context) {
|
||||
console.error("Context is undefined in handleStore for document:", pageId);
|
||||
return;
|
||||
}
|
||||
|
||||
await catchAsync(
|
||||
async () => {
|
||||
if (!state) {
|
||||
handleError(null, {
|
||||
@@ -87,8 +92,12 @@ const handleStore = async ({
|
||||
throw: true,
|
||||
});
|
||||
}
|
||||
let title = "";
|
||||
if (document) {
|
||||
title = extractTextFromHTML(document?.getXmlFragment("title")?.toJSON());
|
||||
}
|
||||
const { documentType } = context as HocusPocusServerContext;
|
||||
const params = requestParameters;
|
||||
|
||||
if (!documentType) {
|
||||
handleError(null, {
|
||||
errorType: "bad-request",
|
||||
@@ -105,12 +114,15 @@ const handleStore = async ({
|
||||
context: context as HocusPocusServerContext,
|
||||
pageId,
|
||||
state,
|
||||
params,
|
||||
title,
|
||||
});
|
||||
},
|
||||
{
|
||||
params: { pageId, documentType: context.documentType },
|
||||
params: {
|
||||
pageId,
|
||||
documentType: context?.documentType || "unknown",
|
||||
},
|
||||
extra: { operation: "store" },
|
||||
}
|
||||
)();
|
||||
};
|
||||
};
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
// hocuspocus extensions and core
|
||||
import { Extension } from "@hocuspocus/server";
|
||||
import { Logger } from "@hocuspocus/extension-logger";
|
||||
import { setupRedisExtension } from "@/core/extensions/redis";
|
||||
import { setupRedisExtension } from "@/core/extensions/setup-redis";
|
||||
import { createDatabaseExtension } from "@/core/extensions/database";
|
||||
import { logger } from "@plane/logger";
|
||||
import { TitleSyncExtension } from "./title-sync";
|
||||
|
||||
export const getExtensions = async (): Promise<Extension[]> => {
|
||||
const extensions: Extension[] = [
|
||||
@@ -14,13 +15,13 @@ export const getExtensions = async (): Promise<Extension[]> => {
|
||||
},
|
||||
}),
|
||||
createDatabaseExtension(),
|
||||
new TitleSyncExtension(),
|
||||
];
|
||||
|
||||
// Add Redis extensions if Redis is available
|
||||
const redisExtension = await setupRedisExtension();
|
||||
if (redisExtension) {
|
||||
logger.info("HocusPocus Redis extension configured ✅");
|
||||
extensions.push(redisExtension);
|
||||
const redisExtensions = await setupRedisExtension();
|
||||
if (redisExtensions) {
|
||||
extensions.push(redisExtensions);
|
||||
}
|
||||
|
||||
return extensions;
|
||||
|
||||
@@ -1,34 +1,17 @@
|
||||
import { Redis as HocusPocusRedis } from "@hocuspocus/extension-redis";
|
||||
// core helpers and utilities
|
||||
import { logger } from "@plane/logger";
|
||||
import { RedisManager } from "@/core/lib/redis-manager";
|
||||
import { Redis } from "@hocuspocus/extension-redis";
|
||||
import { OutgoingMessage } from "@hocuspocus/server";
|
||||
|
||||
/**
|
||||
* Sets up the Redis extension for HocusPocus using the RedisManager singleton
|
||||
* @returns Promise that resolves to a Redis extension array
|
||||
*/
|
||||
export const setupRedisExtension = async () => {
|
||||
const redisManager = RedisManager.getInstance();
|
||||
export class CustomHocuspocusRedisExtension extends Redis {
|
||||
public broadcastToDocument(documentName: string, payload: any): Promise<number> {
|
||||
const stringPayload = typeof payload === "string" ? payload : JSON.stringify(payload);
|
||||
const message = new OutgoingMessage(documentName).writeBroadcastStateless(stringPayload);
|
||||
|
||||
// Wait for Redis connection
|
||||
const redisClient = await redisManager.connect();
|
||||
const emptyPrefix = Buffer.concat([Buffer.from([0])]);
|
||||
|
||||
if (redisClient) {
|
||||
return new HocusPocusRedis({
|
||||
redis: redisClient,
|
||||
});
|
||||
} else {
|
||||
logger.warn(
|
||||
"Redis connection failed, continuing without Redis extension (you won't be able to sync data between multiple plane live servers)"
|
||||
return this.pub.publishBuffer(
|
||||
// we're accessing the private method of the hocuspocus redis extension
|
||||
this["pubKey"](documentName),
|
||||
Buffer.concat([emptyPrefix, Buffer.from(message.toUint8Array())])
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Helper to get the current Redis status
|
||||
* Useful for health checks
|
||||
*/
|
||||
export const getRedisStatus = (): "connected" | "connecting" | "disconnected" | "not-configured" => {
|
||||
const redisManager = RedisManager.getInstance();
|
||||
return redisManager.getStatus();
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
// core helpers and utilities
|
||||
import { RedisManager } from "@/core/lib/redis-manager";
|
||||
import { CustomHocuspocusRedisExtension } from "./redis";
|
||||
import { shutdownManager } from "../shutdown-manager";
|
||||
|
||||
/**
|
||||
* Sets up the Redis extension for HocusPocus using the RedisManager singleton
|
||||
* @returns Promise that resolves to a Redis extension array
|
||||
*/
|
||||
export const setupRedisExtension = async () => {
|
||||
const redisManager = RedisManager.getInstance();
|
||||
|
||||
// Wait for Redis connection
|
||||
const redisClient = await redisManager.connect();
|
||||
|
||||
if (redisClient) {
|
||||
return new CustomHocuspocusRedisExtension({
|
||||
redis: redisClient,
|
||||
});
|
||||
} else {
|
||||
shutdownManager.shutdown("Redis connection failed and could not be recovered", 1);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Helper to get the current Redis status
|
||||
* Useful for health checks
|
||||
*/
|
||||
export const getRedisStatus = (): "connected" | "connecting" | "disconnected" | "not-configured" => {
|
||||
const redisManager = RedisManager.getInstance();
|
||||
return redisManager.getStatus();
|
||||
};
|
||||
@@ -0,0 +1,139 @@
|
||||
// hocuspocus
|
||||
import { Extension, Hocuspocus, Document } from "@hocuspocus/server";
|
||||
import { TiptapTransformer } from "@hocuspocus/transformer";
|
||||
import * as Y from "yjs";
|
||||
// types
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
// editor extensions
|
||||
import { TITLE_EDITOR_EXTENSIONS, createRealtimeEvent } from "@plane/editor";
|
||||
// handlers
|
||||
import { getDocumentHandler } from "@/core/handlers/page-handlers";
|
||||
// helpers
|
||||
import { generateTitleProsemirrorJson } from "@/core/helpers/generate-title-prosemirror-json";
|
||||
import { extractTextFromHTML } from "./title-update/title-utils";
|
||||
import { TitleUpdateManager } from "./title-update/title-update-manager";
|
||||
import { broadcastMessageToPage } from "@/plane-live/lib/broadcast-message-to-page";
|
||||
|
||||
/**
|
||||
* Hocuspocus extension for synchronizing document titles
|
||||
*/
|
||||
export class TitleSyncExtension implements Extension {
|
||||
instance!: Hocuspocus;
|
||||
|
||||
// Maps document names to their observers and update managers
|
||||
private titleObservers: Map<string, (events: Y.YEvent<any>[]) => void> = new Map();
|
||||
private titleUpdateManagers: Map<string, TitleUpdateManager> = new Map();
|
||||
|
||||
/**
|
||||
* Handle document loading - migrate old titles if needed
|
||||
*/
|
||||
async onLoadDocument({ context, document }: { context: HocusPocusServerContext; document: Document }) {
|
||||
try {
|
||||
// initially for on demand migration of old titles to a new title field
|
||||
// in the yjs binary
|
||||
if (document.isEmpty("title")) {
|
||||
const documentHandler = getDocumentHandler(context.documentType);
|
||||
const title = await documentHandler.fetchTitle?.({
|
||||
context,
|
||||
pageId: document.name,
|
||||
});
|
||||
if (title == null) return;
|
||||
const titleField = TiptapTransformer.toYdoc(
|
||||
generateTitleProsemirrorJson(title),
|
||||
"title",
|
||||
// editor
|
||||
TITLE_EDITOR_EXTENSIONS
|
||||
);
|
||||
document.merge(titleField);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error("Error in onLoadDocument: ", error);
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Set up title synchronization for a document after it's loaded
|
||||
*/
|
||||
async afterLoadDocument({
|
||||
document,
|
||||
documentName,
|
||||
context,
|
||||
instance,
|
||||
}: {
|
||||
document: Document;
|
||||
documentName: string;
|
||||
context: HocusPocusServerContext;
|
||||
instance: Hocuspocus;
|
||||
}) {
|
||||
const documentHandler = getDocumentHandler(context.documentType);
|
||||
|
||||
// Create a title update manager for this document
|
||||
const updateManager = new TitleUpdateManager(documentName, context, documentHandler);
|
||||
|
||||
// Store the manager
|
||||
this.titleUpdateManagers.set(documentName, updateManager);
|
||||
|
||||
// Set up observer for title field
|
||||
const titleObserver = (events: Y.YEvent<any>[]) => {
|
||||
let title = "";
|
||||
events.forEach((event) => {
|
||||
title = extractTextFromHTML(event.currentTarget.toJSON());
|
||||
});
|
||||
|
||||
// Schedule an update with the manager
|
||||
const manager = this.titleUpdateManagers.get(documentName);
|
||||
|
||||
// In your titleObserver
|
||||
if (context.parentId) {
|
||||
const event = createRealtimeEvent({
|
||||
user_id: context.userId,
|
||||
workspace_slug: context.workspaceSlug as string,
|
||||
action: "title_updated",
|
||||
page_id: documentName,
|
||||
data: { title },
|
||||
descendants_ids: [],
|
||||
});
|
||||
|
||||
broadcastMessageToPage(instance, context.parentId, event);
|
||||
}
|
||||
|
||||
if (manager) {
|
||||
manager.scheduleUpdate(title);
|
||||
}
|
||||
};
|
||||
|
||||
// Observe the title field
|
||||
document.getXmlFragment("title").observeDeep(titleObserver);
|
||||
this.titleObservers.set(documentName, titleObserver);
|
||||
}
|
||||
|
||||
/**
|
||||
* Force save title before unloading the document
|
||||
*/
|
||||
async beforeUnloadDocument({ documentName }: { documentName: string }) {
|
||||
const updateManager = this.titleUpdateManagers.get(documentName);
|
||||
if (updateManager) {
|
||||
// Force immediate save and wait for it to complete
|
||||
await updateManager.forceSave();
|
||||
// Clean up the manager
|
||||
this.titleUpdateManagers.delete(documentName);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove observers after document unload
|
||||
*/
|
||||
async afterUnloadDocument({ documentName }: { documentName: string }) {
|
||||
// Clean up observer when document is unloaded
|
||||
const observer = this.titleObservers.get(documentName);
|
||||
if (observer) {
|
||||
this.titleObservers.delete(documentName);
|
||||
}
|
||||
|
||||
// Ensure manager is cleaned up if beforeUnloadDocument somehow didn't run
|
||||
if (this.titleUpdateManagers.has(documentName)) {
|
||||
const manager = this.titleUpdateManagers.get(documentName)!;
|
||||
manager.cancel();
|
||||
this.titleUpdateManagers.delete(documentName);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,275 @@
|
||||
/**
|
||||
* DebounceState - Tracks the state of a debounced function
|
||||
*/
|
||||
export interface DebounceState {
|
||||
lastArgs: any[] | null;
|
||||
timerId: ReturnType<typeof setTimeout> | null;
|
||||
lastCallTime: number | undefined;
|
||||
lastExecutionTime: number;
|
||||
inProgress: boolean;
|
||||
abortController: AbortController | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new DebounceState object
|
||||
*/
|
||||
export const createDebounceState = (): DebounceState => ({
|
||||
lastArgs: null,
|
||||
timerId: null,
|
||||
lastCallTime: undefined,
|
||||
lastExecutionTime: 0,
|
||||
inProgress: false,
|
||||
abortController: null,
|
||||
});
|
||||
|
||||
/**
|
||||
* DebounceOptions - Configuration options for debounce
|
||||
*/
|
||||
export interface DebounceOptions {
|
||||
/** The wait time in milliseconds */
|
||||
wait: number;
|
||||
|
||||
/** Optional logging prefix for debug messages */
|
||||
logPrefix?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enhanced debounce manager with abort support
|
||||
* Manages the state and timing of debounced function calls
|
||||
*/
|
||||
export class DebounceManager {
|
||||
private state: DebounceState;
|
||||
private wait: number;
|
||||
private logPrefix: string;
|
||||
|
||||
/**
|
||||
* Creates a new DebounceManager
|
||||
* @param options Debounce configuration options
|
||||
*/
|
||||
constructor(options: DebounceOptions) {
|
||||
this.state = createDebounceState();
|
||||
this.wait = options.wait;
|
||||
this.logPrefix = options.logPrefix || "";
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule a debounced function call
|
||||
* @param func The function to call
|
||||
* @param args The arguments to pass to the function
|
||||
*/
|
||||
schedule(func: (...args: any[]) => Promise<void>, ...args: any[]): void {
|
||||
// Always update the last arguments
|
||||
this.state.lastArgs = args;
|
||||
|
||||
const time = Date.now();
|
||||
this.state.lastCallTime = time;
|
||||
|
||||
// If an operation is in progress, just store the new args and start the timer
|
||||
if (this.state.inProgress) {
|
||||
// Always restart the timer for the new call, even if an operation is in progress
|
||||
if (this.state.timerId) {
|
||||
clearTimeout(this.state.timerId);
|
||||
}
|
||||
|
||||
this.state.timerId = setTimeout(() => {
|
||||
this.timerExpired(func);
|
||||
}, this.wait);
|
||||
return;
|
||||
}
|
||||
|
||||
// If already scheduled, update the args and restart the timer
|
||||
if (this.state.timerId) {
|
||||
clearTimeout(this.state.timerId);
|
||||
this.state.timerId = setTimeout(() => {
|
||||
this.timerExpired(func);
|
||||
}, this.wait);
|
||||
return;
|
||||
}
|
||||
|
||||
// Start the timer for the trailing edge execution
|
||||
this.state.timerId = setTimeout(() => {
|
||||
this.timerExpired(func);
|
||||
}, this.wait);
|
||||
}
|
||||
|
||||
/**
|
||||
* Called when the timer expires
|
||||
*/
|
||||
private timerExpired(func: (...args: any[]) => Promise<void>): void {
|
||||
const time = Date.now();
|
||||
|
||||
// Check if this timer expiration represents the end of the debounce period
|
||||
if (this.shouldInvoke(time)) {
|
||||
// Execute the function
|
||||
this.executeFunction(func, time);
|
||||
return;
|
||||
}
|
||||
|
||||
// Otherwise restart the timer
|
||||
this.state.timerId = setTimeout(() => {
|
||||
this.timerExpired(func);
|
||||
}, this.remainingWait(time));
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the debounced function
|
||||
*/
|
||||
private executeFunction(func: (...args: any[]) => Promise<void>, time: number): void {
|
||||
this.state.timerId = null;
|
||||
this.state.lastExecutionTime = time;
|
||||
|
||||
// Execute the function asynchronously
|
||||
this.performFunction(func).catch((error) => {
|
||||
console.error(`${this.logPrefix}: Error in execution:`, error);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform the actual function call, handling any in-progress operations
|
||||
*/
|
||||
private async performFunction(func: (...args: any[]) => Promise<void>): Promise<void> {
|
||||
const args = this.state.lastArgs;
|
||||
if (!args) return;
|
||||
|
||||
// Store the args we're about to use
|
||||
const currentArgs = [...args];
|
||||
|
||||
// If another operation is in progress, abort it
|
||||
await this.abortOngoingOperation();
|
||||
|
||||
// Mark that we're starting a new operation
|
||||
this.state.inProgress = true;
|
||||
this.state.abortController = new AbortController();
|
||||
|
||||
try {
|
||||
// Add the abort signal to the arguments if the function can use it
|
||||
const execArgs = [...currentArgs];
|
||||
execArgs.push(this.state.abortController.signal);
|
||||
|
||||
await func(...execArgs);
|
||||
|
||||
// Only clear lastArgs if they haven't been changed during this operation
|
||||
if (this.state.lastArgs && this.arraysEqual(this.state.lastArgs, currentArgs)) {
|
||||
this.state.lastArgs = null;
|
||||
|
||||
// Clear any timer as we've successfully processed the latest args
|
||||
if (this.state.timerId) {
|
||||
clearTimeout(this.state.timerId);
|
||||
this.state.timerId = null;
|
||||
}
|
||||
} else if (this.state.lastArgs) {
|
||||
// If lastArgs have changed during this operation, the timer should already be running
|
||||
// but let's make sure it is
|
||||
if (!this.state.timerId) {
|
||||
this.state.timerId = setTimeout(() => {
|
||||
this.timerExpired(func);
|
||||
}, this.wait);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof Error && error.name === "AbortError") {
|
||||
// Nothing to do here, the new operation will be triggered by the timer expiration
|
||||
} else {
|
||||
console.error(`${this.logPrefix}: Error during operation:`, error);
|
||||
|
||||
// On error (not abort), make sure we have a timer running to retry
|
||||
if (!this.state.timerId && this.state.lastArgs) {
|
||||
this.state.timerId = setTimeout(() => {
|
||||
this.timerExpired(func);
|
||||
}, this.wait);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.state.inProgress = false;
|
||||
this.state.abortController = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Abort any ongoing operation
|
||||
*/
|
||||
private async abortOngoingOperation(): Promise<void> {
|
||||
if (this.state.inProgress && this.state.abortController) {
|
||||
this.state.abortController.abort();
|
||||
|
||||
// Small delay to ensure the abort has had time to propagate
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
|
||||
// Double-check that state has been reset, force it if not
|
||||
if (this.state.inProgress || this.state.abortController) {
|
||||
this.state.inProgress = false;
|
||||
this.state.abortController = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if we should invoke the function now
|
||||
*/
|
||||
private shouldInvoke(time: number): boolean {
|
||||
// Either this is the first call, or we've waited long enough since the last call
|
||||
return this.state.lastCallTime === undefined || time - this.state.lastCallTime >= this.wait;
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate how much longer we should wait
|
||||
*/
|
||||
private remainingWait(time: number): number {
|
||||
const timeSinceLastCall = time - (this.state.lastCallTime || 0);
|
||||
return Math.max(0, this.wait - timeSinceLastCall);
|
||||
}
|
||||
|
||||
/**
|
||||
* Force immediate execution
|
||||
*/
|
||||
async flush(func: (...args: any[]) => Promise<void>): Promise<void> {
|
||||
// Clear any pending timeout
|
||||
if (this.state.timerId) {
|
||||
clearTimeout(this.state.timerId);
|
||||
this.state.timerId = null;
|
||||
}
|
||||
|
||||
// Reset timing state
|
||||
this.state.lastCallTime = undefined;
|
||||
|
||||
// Perform the function immediately
|
||||
if (this.state.lastArgs) {
|
||||
await this.performFunction(func);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel any pending operations without executing
|
||||
*/
|
||||
cancel(): void {
|
||||
// Clear any pending timeout
|
||||
if (this.state.timerId) {
|
||||
clearTimeout(this.state.timerId);
|
||||
this.state.timerId = null;
|
||||
}
|
||||
|
||||
// Reset timing state
|
||||
this.state.lastCallTime = undefined;
|
||||
|
||||
// Abort any in-progress operation
|
||||
if (this.state.inProgress && this.state.abortController) {
|
||||
this.state.abortController.abort();
|
||||
this.state.inProgress = false;
|
||||
this.state.abortController = null;
|
||||
}
|
||||
|
||||
// Clear args
|
||||
this.state.lastArgs = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare two arrays for equality
|
||||
*/
|
||||
private arraysEqual(a: any[], b: any[]): boolean {
|
||||
if (a.length !== b.length) return false;
|
||||
for (let i = 0; i < a.length; i++) {
|
||||
if (a[i] !== b[i]) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
import { DocumentHandler } from "@/core/types/document-handler";
|
||||
import { DebounceManager } from "./debounce";
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
import { logger } from "@plane/logger";
|
||||
|
||||
/**
|
||||
* Manages title update operations for a single document
|
||||
* Handles debouncing, aborting, and force saving title updates
|
||||
*/
|
||||
export class TitleUpdateManager {
|
||||
private documentName: string;
|
||||
private context: HocusPocusServerContext;
|
||||
private documentHandler: DocumentHandler;
|
||||
private debounceManager: DebounceManager;
|
||||
private lastTitle: string | null = null;
|
||||
|
||||
/**
|
||||
* Create a new TitleUpdateManager instance
|
||||
*/
|
||||
constructor(
|
||||
documentName: string,
|
||||
context: HocusPocusServerContext,
|
||||
documentHandler: DocumentHandler,
|
||||
wait: number = 5000
|
||||
) {
|
||||
this.documentName = documentName;
|
||||
this.context = context;
|
||||
this.documentHandler = documentHandler;
|
||||
|
||||
// Set up debounce manager with logging
|
||||
this.debounceManager = new DebounceManager({
|
||||
wait,
|
||||
logPrefix: `TitleManager[${documentName.substring(0, 8)}]`,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule a debounced title update
|
||||
*/
|
||||
scheduleUpdate(title: string): void {
|
||||
// Store the latest title
|
||||
this.lastTitle = title;
|
||||
|
||||
// Schedule the update with the debounce manager
|
||||
this.debounceManager.schedule(this.updateTitle.bind(this), title);
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the title - will be called by the debounce manager
|
||||
*/
|
||||
private async updateTitle(title: string, signal?: AbortSignal): Promise<void> {
|
||||
if (!this.documentHandler.updateTitle) {
|
||||
logger.warn(`No updateTitle method found for document ${this.documentName}`);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.documentHandler.updateTitle({
|
||||
context: this.context,
|
||||
pageId: this.documentName,
|
||||
title: title,
|
||||
abortSignal: signal,
|
||||
});
|
||||
|
||||
// Clear last title only if it matches what we just updated
|
||||
if (this.lastTitle === title) {
|
||||
this.lastTitle = null;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Error updating title for ${this.documentName}:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Force save the current title immediately
|
||||
*/
|
||||
async forceSave(): Promise<void> {
|
||||
// Ensure we have the current title
|
||||
if (!this.lastTitle) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Use the debounce manager to flush the operation
|
||||
await this.debounceManager.flush(this.updateTitle.bind(this));
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel any pending updates
|
||||
*/
|
||||
cancel(): void {
|
||||
this.debounceManager.cancel();
|
||||
this.lastTitle = null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
/**
|
||||
* Utility function to extract text from HTML content
|
||||
*/
|
||||
export const extractTextFromHTML = (html: string): string => {
|
||||
// Use a regex to extract text between tags
|
||||
const textMatch = html.replace(/<[^>]*>/g, "");
|
||||
return textMatch || "";
|
||||
};
|
||||
@@ -0,0 +1,29 @@
|
||||
import { Request, Response } from "express";
|
||||
import { manualLogger } from "@/core/helpers/logger";
|
||||
import { TConvertDocumentRequestBody } from "@/core/types/common";
|
||||
import { convertHTMLDocumentToAllFormats } from "@plane/editor";
|
||||
|
||||
export const handleConvertDocument = (req: Request, res: Response) => {
|
||||
const { description_html, variant } = req.body as TConvertDocumentRequestBody;
|
||||
try {
|
||||
if (description_html === undefined || variant === undefined) {
|
||||
res.status(400).send({
|
||||
message: "Missing required fields",
|
||||
});
|
||||
return;
|
||||
}
|
||||
const { description, description_binary } = convertHTMLDocumentToAllFormats({
|
||||
document_html: description_html,
|
||||
variant,
|
||||
});
|
||||
res.status(200).json({
|
||||
description,
|
||||
description_binary,
|
||||
});
|
||||
} catch (error) {
|
||||
manualLogger.error("Error in /convert-document endpoint:", error);
|
||||
res.status(500).send({
|
||||
message: `Internal server error. ${error}`,
|
||||
});
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1 @@
|
||||
export { handleConvertDocument } from "./convert-document.handler";
|
||||
+9
-7
@@ -1,11 +1,12 @@
|
||||
import { DocumentHandler, HandlerContext, HandlerDefinition } from "@/core/types/document-handler";
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
import { DocumentHandler, HandlerDefinition } from "@/core/types/document-handler";
|
||||
|
||||
/**
|
||||
* Class that manages handler selection based on multiple criteria
|
||||
*/
|
||||
export class DocumentHandlerFactory {
|
||||
private handlers: HandlerDefinition[] = [];
|
||||
|
||||
|
||||
/**
|
||||
* Register a handler with its selection criteria
|
||||
*/
|
||||
@@ -14,14 +15,15 @@ export class DocumentHandlerFactory {
|
||||
// Sort handlers by priority (highest first)
|
||||
this.handlers.sort((a, b) => b.priority - a.priority);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get the appropriate handler based on the provided context
|
||||
*/
|
||||
getHandler(context: HandlerContext): DocumentHandler {
|
||||
getHandler(context: Partial<HocusPocusServerContext>): DocumentHandler {
|
||||
// Find the first handler whose selector returns true
|
||||
const matchingHandler = this.handlers.find(h => h.selector(context));
|
||||
|
||||
const matchingHandler = this.handlers.find((h) => h.selector(context));
|
||||
// console.log("matchingHandler:", matchingHandler);
|
||||
|
||||
// Return the matching handler or fall back to null/undefined
|
||||
// (This will cause an error if no handlers match, which is good for debugging)
|
||||
return matchingHandler?.handler as DocumentHandler;
|
||||
@@ -29,4 +31,4 @@ export class DocumentHandlerFactory {
|
||||
}
|
||||
|
||||
// Create the singleton instance
|
||||
export const handlerFactory = new DocumentHandlerFactory();
|
||||
export const handlerFactory = new DocumentHandlerFactory();
|
||||
+7
-12
@@ -1,12 +1,7 @@
|
||||
import { DocumentHandler, HandlerContext } from "@/core/types/document-handler";
|
||||
import { handlerFactory } from "@/core/handlers/document-handlers/handler-factory";
|
||||
|
||||
import { HocusPocusServerContext } from "@/core/types/common";
|
||||
import { initializeDocumentHandlers } from "@/plane-live/document-types";
|
||||
|
||||
// Initialize all CE document handlers
|
||||
initializeDocumentHandlers();
|
||||
import { DocumentHandler } from "@/core/types/document-handler";
|
||||
import { handlerFactory } from "@/core/handlers/page-handlers/handler-factory";
|
||||
|
||||
import { HocusPocusServerContext, TDocumentTypes } from "@/core/types/common";
|
||||
/**
|
||||
* Get a document handler based on the provided context criteria
|
||||
* @param documentType The primary document type
|
||||
@@ -14,12 +9,12 @@ initializeDocumentHandlers();
|
||||
* @returns The appropriate document handler
|
||||
*/
|
||||
export function getDocumentHandler(
|
||||
documentType: string,
|
||||
additionalContext: Omit<HocusPocusServerContext, "documentType">
|
||||
documentType: TDocumentTypes,
|
||||
additionalContext?: Omit<HocusPocusServerContext, "documentType">
|
||||
): DocumentHandler {
|
||||
// Create a context object with all criteria
|
||||
const context: HandlerContext = {
|
||||
documentType: documentType as any,
|
||||
const context: Partial<HocusPocusServerContext> = {
|
||||
documentType: documentType,
|
||||
...additionalContext,
|
||||
};
|
||||
|
||||
@@ -1,44 +0,0 @@
|
||||
// plane editor
|
||||
import {
|
||||
getAllDocumentFormatsFromDocumentEditorBinaryData,
|
||||
getAllDocumentFormatsFromRichTextEditorBinaryData,
|
||||
getBinaryDataFromDocumentEditorHTMLString,
|
||||
getBinaryDataFromRichTextEditorHTMLString,
|
||||
} from "@plane/editor";
|
||||
// plane types
|
||||
import { TDocumentPayload } from "@plane/types";
|
||||
|
||||
type TArgs = {
|
||||
document_html: string;
|
||||
variant: "rich" | "document";
|
||||
};
|
||||
|
||||
export const convertHTMLDocumentToAllFormats = (args: TArgs): TDocumentPayload => {
|
||||
const { document_html, variant } = args;
|
||||
|
||||
let allFormats: TDocumentPayload;
|
||||
|
||||
if (variant === "rich") {
|
||||
const contentBinary = getBinaryDataFromRichTextEditorHTMLString(document_html);
|
||||
const { contentBinaryEncoded, contentHTML, contentJSON } =
|
||||
getAllDocumentFormatsFromRichTextEditorBinaryData(contentBinary);
|
||||
allFormats = {
|
||||
description: contentJSON,
|
||||
description_html: contentHTML,
|
||||
description_binary: contentBinaryEncoded,
|
||||
};
|
||||
} else if (variant === "document") {
|
||||
const contentBinary = getBinaryDataFromDocumentEditorHTMLString(document_html);
|
||||
const { contentBinaryEncoded, contentHTML, contentJSON } =
|
||||
getAllDocumentFormatsFromDocumentEditorBinaryData(contentBinary);
|
||||
allFormats = {
|
||||
description: contentJSON,
|
||||
description_html: contentHTML,
|
||||
description_binary: contentBinaryEncoded,
|
||||
};
|
||||
} else {
|
||||
throw new Error(`Invalid variant provided: ${variant}`);
|
||||
}
|
||||
|
||||
return allFormats;
|
||||
};
|
||||
@@ -0,0 +1,21 @@
|
||||
export const generateTitleProsemirrorJson = (text: string) => {
|
||||
return {
|
||||
type: "doc",
|
||||
content: [
|
||||
{
|
||||
type: "heading",
|
||||
attrs: { level: 1 },
|
||||
...(text
|
||||
? {
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text,
|
||||
},
|
||||
],
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
],
|
||||
};
|
||||
};
|
||||
@@ -1,20 +1,34 @@
|
||||
import { getSchema } from "@tiptap/core";
|
||||
import { generateHTML, generateJSON } from "@tiptap/html";
|
||||
import { prosemirrorJSONToYDoc, yXmlFragmentToProseMirrorRootNode } from "y-prosemirror";
|
||||
import * as Y from "yjs"
|
||||
import * as Y from "yjs";
|
||||
// plane editor
|
||||
import { CoreEditorExtensionsWithoutProps, DocumentEditorExtensionsWithoutProps } from "@plane/editor/lib";
|
||||
|
||||
const DOCUMENT_EDITOR_EXTENSIONS = [
|
||||
export const DOCUMENT_EDITOR_EXTENSIONS = [
|
||||
...CoreEditorExtensionsWithoutProps,
|
||||
...DocumentEditorExtensionsWithoutProps,
|
||||
];
|
||||
const documentEditorSchema = getSchema(DOCUMENT_EDITOR_EXTENSIONS);
|
||||
export const documentEditorSchema = getSchema(DOCUMENT_EDITOR_EXTENSIONS);
|
||||
|
||||
export const getAllDocumentFormatsFromBinaryData = (description: Uint8Array): {
|
||||
/**
|
||||
* Extracts the text content from an HTML string
|
||||
* @param html HTML string
|
||||
* @returns text content
|
||||
*/
|
||||
export const extractTextFromHTML = (html: string): string => {
|
||||
// Use a regex to extract text between tags
|
||||
const textMatch = html.replace(/<[^>]*>/g, "");
|
||||
return textMatch || "";
|
||||
};
|
||||
|
||||
export const getAllDocumentFormatsFromBinaryData = (
|
||||
description: Uint8Array
|
||||
): {
|
||||
contentBinaryEncoded: string;
|
||||
contentJSON: object;
|
||||
contentHTML: string;
|
||||
titleHTML: string;
|
||||
} => {
|
||||
// encode binary description data
|
||||
const base64Data = Buffer.from(description).toString("base64");
|
||||
@@ -22,10 +36,11 @@ export const getAllDocumentFormatsFromBinaryData = (description: Uint8Array): {
|
||||
Y.applyUpdate(yDoc, description);
|
||||
// convert to JSON
|
||||
const type = yDoc.getXmlFragment("default");
|
||||
const contentJSON = yXmlFragmentToProseMirrorRootNode(
|
||||
type,
|
||||
documentEditorSchema
|
||||
).toJSON();
|
||||
const contentJSON = yXmlFragmentToProseMirrorRootNode(type, documentEditorSchema).toJSON();
|
||||
|
||||
const title = yDoc.getXmlFragment("title");
|
||||
const titleJSON = yXmlFragmentToProseMirrorRootNode(title, documentEditorSchema).toJSON();
|
||||
const titleHTML = extractTextFromHTML(generateHTML(titleJSON, DOCUMENT_EDITOR_EXTENSIONS));
|
||||
// convert to HTML
|
||||
const contentHTML = generateHTML(contentJSON, DOCUMENT_EDITOR_EXTENSIONS);
|
||||
|
||||
@@ -33,27 +48,24 @@ export const getAllDocumentFormatsFromBinaryData = (description: Uint8Array): {
|
||||
contentBinaryEncoded: base64Data,
|
||||
contentJSON,
|
||||
contentHTML,
|
||||
titleHTML,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
export const getBinaryDataFromHTMLString = (descriptionHTML: string): {
|
||||
contentBinary: Uint8Array
|
||||
export const getBinaryDataFromHTMLString = (
|
||||
descriptionHTML: string
|
||||
): {
|
||||
contentBinary: Uint8Array;
|
||||
} => {
|
||||
// convert HTML to JSON
|
||||
const contentJSON = generateJSON(
|
||||
descriptionHTML ?? "<p></p>",
|
||||
DOCUMENT_EDITOR_EXTENSIONS
|
||||
);
|
||||
const contentJSON = generateJSON(descriptionHTML ?? "<p></p>", DOCUMENT_EDITOR_EXTENSIONS);
|
||||
// convert JSON to Y.Doc format
|
||||
const transformedData = prosemirrorJSONToYDoc(
|
||||
documentEditorSchema,
|
||||
contentJSON,
|
||||
"default"
|
||||
);
|
||||
const transformedData = prosemirrorJSONToYDoc(documentEditorSchema, contentJSON, "default");
|
||||
// convert Y.Doc to Uint8Array format
|
||||
const encodedData = Y.encodeStateAsUpdate(transformedData);
|
||||
|
||||
return {
|
||||
contentBinary: encodedData
|
||||
}
|
||||
}
|
||||
contentBinary: encodedData,
|
||||
};
|
||||
};
|
||||
|
||||
|
||||
@@ -1,261 +0,0 @@
|
||||
import { handleError } from "./error-handling/error-factory";
|
||||
|
||||
/**
|
||||
* A simple validation utility that integrates with our error system.
|
||||
*
|
||||
* This provides a fluent interface for validating data and throwing
|
||||
* appropriate errors if validation fails.
|
||||
*/
|
||||
export class Validator<T> {
|
||||
constructor(
|
||||
private readonly data: T,
|
||||
private readonly name: string = "data"
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Ensures a value is defined (not undefined or null)
|
||||
*/
|
||||
required(message?: string): Validator<T> {
|
||||
if (this.data === undefined || this.data === null) {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} is required`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateRequired',
|
||||
extraContext: { field: this.name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensures a value is a string
|
||||
*/
|
||||
string(message?: string): Validator<T> {
|
||||
if (typeof this.data !== "string") {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} must be a string`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateString',
|
||||
extraContext: { field: this.name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensures a string is not empty
|
||||
*/
|
||||
notEmpty(message?: string): Validator<T> {
|
||||
if (typeof this.data === "string" && this.data.trim() === "") {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} cannot be empty`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateNonEmptyString',
|
||||
extraContext: { field: this.name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensures a value is a number
|
||||
*/
|
||||
number(message?: string): Validator<T> {
|
||||
if (typeof this.data !== "number" || isNaN(this.data)) {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} must be a valid number`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateNumber',
|
||||
extraContext: { field: this.name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensures an array is not empty
|
||||
*/
|
||||
nonEmptyArray(message?: string): Validator<T> {
|
||||
if (!Array.isArray(this.data) || this.data.length === 0) {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} must be a non-empty array`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateArray',
|
||||
extraContext: { field: this.name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensures a value matches a regular expression
|
||||
*/
|
||||
match(regex: RegExp, message?: string): Validator<T> {
|
||||
if (typeof this.data !== "string" || !regex.test(this.data)) {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} has an invalid format`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateFormat',
|
||||
extraContext: { field: this.name, format: regex.toString() },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensures a value is one of the allowed values
|
||||
*/
|
||||
oneOf(allowedValues: any[], message?: string): Validator<T> {
|
||||
if (!allowedValues.includes(this.data)) {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} must be one of: ${allowedValues.join(", ")}`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateEnum',
|
||||
extraContext: { field: this.name, allowedValues },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Custom validation function
|
||||
*/
|
||||
custom(validationFn: (value: T) => boolean, message?: string): Validator<T> {
|
||||
if (!validationFn(this.data)) {
|
||||
throw handleError(new ValidationError(this.name, message || `${this.name} is invalid`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateCustom',
|
||||
extraContext: { field: this.name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the validated data
|
||||
*/
|
||||
get(): T {
|
||||
return this.data;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new validator for a value
|
||||
*/
|
||||
export const validate = <T>(data: T, name?: string): Validator<T> => {
|
||||
return new Validator(data, name);
|
||||
};
|
||||
|
||||
export default validate;
|
||||
|
||||
export class ValidationError extends Error {
|
||||
constructor(public name: string, message: string) {
|
||||
super(message);
|
||||
this.name = name;
|
||||
}
|
||||
}
|
||||
|
||||
export const validateRequired = (value: any, name: string, message?: string) => {
|
||||
if (value === undefined || value === null) {
|
||||
throw handleError(new ValidationError(name, message || `${name} is required`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateRequired',
|
||||
extraContext: { field: name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateString = (value: any, name: string, message?: string) => {
|
||||
if (typeof value !== 'string') {
|
||||
throw handleError(new ValidationError(name, message || `${name} must be a string`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateString',
|
||||
extraContext: { field: name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateNonEmptyString = (value: string, name: string, message?: string) => {
|
||||
if (!value.trim()) {
|
||||
throw handleError(new ValidationError(name, message || `${name} cannot be empty`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateNonEmptyString',
|
||||
extraContext: { field: name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateNumber = (value: any, name: string, message?: string) => {
|
||||
if (typeof value !== 'number' || isNaN(value)) {
|
||||
throw handleError(new ValidationError(name, message || `${name} must be a valid number`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateNumber',
|
||||
extraContext: { field: name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateArray = (value: any, name: string, message?: string) => {
|
||||
if (!Array.isArray(value) || value.length === 0) {
|
||||
throw handleError(new ValidationError(name, message || `${name} must be a non-empty array`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateArray',
|
||||
extraContext: { field: name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateFormat = (value: string, name: string, format: RegExp, message?: string) => {
|
||||
if (!format.test(value)) {
|
||||
throw handleError(new ValidationError(name, message || `${name} has an invalid format`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateFormat',
|
||||
extraContext: { field: name, format: format.toString() },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateEnum = (value: any, name: string, allowedValues: any[], message?: string) => {
|
||||
if (!allowedValues.includes(value)) {
|
||||
throw handleError(new ValidationError(name, message || `${name} must be one of: ${allowedValues.join(", ")}`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateEnum',
|
||||
extraContext: { field: name, allowedValues },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
export const validateCustom = (value: any, name: string, validator: (value: any) => boolean, message?: string) => {
|
||||
if (!validator(value)) {
|
||||
throw handleError(new ValidationError(name, message || `${name} is invalid`), {
|
||||
errorType: 'bad-request',
|
||||
component: 'validation',
|
||||
operation: 'validateCustom',
|
||||
extraContext: { field: name },
|
||||
throw: true
|
||||
});
|
||||
}
|
||||
};
|
||||
@@ -7,48 +7,44 @@ import { handleAuthentication } from "@/core/lib/authentication";
|
||||
import { getExtensions } from "@/core/extensions/index";
|
||||
import { DocumentCollaborativeEvents, TDocumentEventsServer } from "@plane/editor/lib";
|
||||
// editor types
|
||||
import { TUserDetails } from "@plane/editor";
|
||||
import { EventToPayloadMap, TUserDetails, createRealtimeEvent } from "@plane/editor";
|
||||
// types
|
||||
import { TDocumentTypes, type HocusPocusServerContext } from "@/core/types/common";
|
||||
// error handling
|
||||
import { catchAsync } from "@/core/helpers/error-handling/error-handler";
|
||||
import { handleError } from "@/core/helpers/error-handling/error-factory";
|
||||
// server agent
|
||||
import { serverAgentManager } from "./agents/server-agent";
|
||||
|
||||
export const getHocusPocusServer = async () => {
|
||||
const extensions = await getExtensions();
|
||||
const serverName = process.env.HOSTNAME || uuidv4();
|
||||
return Server.configure({
|
||||
const server = Server.configure({
|
||||
name: serverName,
|
||||
onAuthenticate: async ({
|
||||
requestHeaders,
|
||||
context,
|
||||
requestParameters,
|
||||
// user id used as token for authentication
|
||||
context,
|
||||
token,
|
||||
}: {
|
||||
requestHeaders: IncomingHttpHeaders;
|
||||
context: HocusPocusServerContext; // Better than 'any', still allows property assignment
|
||||
context: HocusPocusServerContext;
|
||||
requestParameters: URLSearchParams;
|
||||
token: string;
|
||||
}) => {
|
||||
// need to rethrow all errors since hocuspocus needs to know to stop
|
||||
// further propagation of events to other document lifecycle
|
||||
return catchAsync(
|
||||
async () => {
|
||||
let cookie: string | undefined = undefined;
|
||||
let userId: string | undefined = undefined;
|
||||
|
||||
// Extract cookie (fallback to request headers) and userId from token (for scenarios where
|
||||
// the cookies are not passed in the request headers)
|
||||
// Extract cookie (fallback to request headers) and userId from token
|
||||
try {
|
||||
const parsedToken = JSON.parse(token) as TUserDetails;
|
||||
userId = parsedToken.id;
|
||||
cookie = parsedToken.cookie;
|
||||
} catch (error) {
|
||||
// If token parsing fails, fallback to request headers
|
||||
console.error("Token parsing failed, using request headers:", error);
|
||||
} finally {
|
||||
// If cookie is still not found, fallback to request headers
|
||||
if (!cookie) {
|
||||
cookie = requestHeaders.cookie?.toString();
|
||||
}
|
||||
@@ -66,9 +62,12 @@ export const getHocusPocusServer = async () => {
|
||||
}
|
||||
|
||||
context.documentType = requestParameters.get("documentType")?.toString() as TDocumentTypes;
|
||||
context.cookie = cookie ?? requestParameters.get("cookie");
|
||||
context.cookie = cookie ?? requestParameters.get("cookie") ?? "";
|
||||
context.userId = userId;
|
||||
context.workspaceSlug = requestParameters.get("workspaceSlug")?.toString() as string;
|
||||
context.workspaceSlug = requestParameters.get("workspaceSlug")?.toString() ?? "";
|
||||
context.parentId = requestParameters.get("parentPageId")?.toString() ?? undefined;
|
||||
context.projectId = requestParameters.get("projectId")?.toString() ?? "";
|
||||
context.teamspaceId = requestParameters.get("teamspaceId")?.toString() ?? "";
|
||||
|
||||
return await handleAuthentication({
|
||||
cookie: context.cookie,
|
||||
@@ -77,24 +76,76 @@ export const getHocusPocusServer = async () => {
|
||||
});
|
||||
},
|
||||
{ extra: { operation: "authenticate" } },
|
||||
{
|
||||
rethrow: true,
|
||||
}
|
||||
{ rethrow: true }
|
||||
)();
|
||||
},
|
||||
onStateless: async ({ payload, document }) => {
|
||||
onStateless: async ({ payload, document, connection }) => {
|
||||
return catchAsync(
|
||||
async () => {
|
||||
// broadcast the client event (derived from the server event) to all the clients so that they can update their state
|
||||
const response = DocumentCollaborativeEvents[payload as TDocumentEventsServer].client;
|
||||
if (response) {
|
||||
document.broadcastStateless(response);
|
||||
const payloadStr = payload as string;
|
||||
|
||||
// Function to safely parse JSON without throwing exceptions
|
||||
const safeJsonParse = (str: string) => {
|
||||
try {
|
||||
return { success: true, data: JSON.parse(str) };
|
||||
} catch (e) {
|
||||
return { success: false, error: e };
|
||||
}
|
||||
};
|
||||
|
||||
// First check if this is a known document event
|
||||
const documentEvent = DocumentCollaborativeEvents[payload as TDocumentEventsServer]?.client;
|
||||
|
||||
if (documentEvent) {
|
||||
const eventType = documentEvent as keyof EventToPayloadMap;
|
||||
|
||||
let eventData: Partial<EventToPayloadMap[typeof eventType]> = {
|
||||
user_id: connection.context.userId,
|
||||
};
|
||||
|
||||
if (eventType === "archived") {
|
||||
eventData = {
|
||||
...eventData,
|
||||
archived_at: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
const realtimeEvent = createRealtimeEvent({
|
||||
action: eventType,
|
||||
page_id: document.name,
|
||||
descendants_ids: [],
|
||||
data: eventData as EventToPayloadMap[typeof eventType],
|
||||
workspace_slug: connection.context.workspaceSlug || "",
|
||||
user_id: connection.context.userId || "",
|
||||
});
|
||||
|
||||
// Broadcast the event
|
||||
document.broadcastStateless(JSON.stringify(realtimeEvent));
|
||||
return;
|
||||
}
|
||||
|
||||
// If not a document event, try to parse as JSON
|
||||
const parseResult = safeJsonParse(payloadStr);
|
||||
|
||||
if (parseResult.success && parseResult.data && typeof parseResult.data === "object") {
|
||||
const parsedPayload = parseResult.data as {
|
||||
workspaceSlug?: string;
|
||||
projectId?: string;
|
||||
action?: string;
|
||||
};
|
||||
|
||||
// Handle synced action
|
||||
if (parsedPayload.action === "synced" && parsedPayload.workspaceSlug) {
|
||||
serverAgentManager.notifySyncTrigger(document.name, connection.context);
|
||||
return;
|
||||
}
|
||||
}
|
||||
},
|
||||
{ extra: { operation: "stateless", payload } }
|
||||
);
|
||||
)();
|
||||
},
|
||||
extensions,
|
||||
debounce: 1000,
|
||||
extensions: [...extensions],
|
||||
debounce: 10000,
|
||||
});
|
||||
return server;
|
||||
};
|
||||
|
||||
@@ -17,7 +17,6 @@ export const handleAuthentication = async (props: Props) => {
|
||||
try {
|
||||
response = await userService.currentUser(cookie);
|
||||
} catch (error) {
|
||||
console.log("caught?");
|
||||
handleError(error, {
|
||||
errorType: "unauthorized",
|
||||
message: "Failed to authenticate user",
|
||||
|
||||
@@ -1,64 +0,0 @@
|
||||
// helpers
|
||||
import { getAllDocumentFormatsFromBinaryData, getBinaryDataFromHTMLString } from "@/core/helpers/page";
|
||||
// services
|
||||
import { PageService } from "@/core/services/page.service";
|
||||
const pageService = new PageService();
|
||||
|
||||
export const updatePageDescription = async (
|
||||
params: URLSearchParams,
|
||||
pageId: string,
|
||||
updatedDescription: Uint8Array,
|
||||
cookie: string | undefined
|
||||
) => {
|
||||
if (!(updatedDescription instanceof Uint8Array)) {
|
||||
throw new Error("Invalid updatedDescription: must be an instance of Uint8Array");
|
||||
}
|
||||
|
||||
const workspaceSlug = params.get("workspaceSlug")?.toString();
|
||||
const projectId = params.get("projectId")?.toString();
|
||||
if (!workspaceSlug || !projectId || !cookie) return;
|
||||
|
||||
const { contentBinaryEncoded, contentHTML, contentJSON } = getAllDocumentFormatsFromBinaryData(updatedDescription);
|
||||
const payload = {
|
||||
description_binary: contentBinaryEncoded,
|
||||
description_html: contentHTML,
|
||||
description: contentJSON,
|
||||
};
|
||||
|
||||
await pageService.updateDescription(workspaceSlug, projectId, pageId, payload, cookie);
|
||||
};
|
||||
|
||||
const fetchDescriptionHTMLAndTransform = async (
|
||||
workspaceSlug: string,
|
||||
projectId: string,
|
||||
pageId: string,
|
||||
cookie: string
|
||||
) => {
|
||||
if (!workspaceSlug || !projectId || !cookie) return;
|
||||
|
||||
const pageDetails = await pageService.fetchDetails(workspaceSlug, projectId, pageId, cookie);
|
||||
const { contentBinary } = getBinaryDataFromHTMLString(pageDetails.description_html ?? "<p></p>");
|
||||
return contentBinary;
|
||||
};
|
||||
|
||||
export const fetchPageDescriptionBinary = async (
|
||||
params: URLSearchParams,
|
||||
pageId: string,
|
||||
cookie: string | undefined
|
||||
) => {
|
||||
const workspaceSlug = params.get("workspaceSlug")?.toString();
|
||||
const projectId = params.get("projectId")?.toString();
|
||||
if (!workspaceSlug || !projectId || !cookie) return null;
|
||||
|
||||
const response = await pageService.fetchDescriptionBinary(workspaceSlug, projectId, pageId, cookie);
|
||||
const binaryData = new Uint8Array(response);
|
||||
|
||||
if (binaryData.byteLength === 0) {
|
||||
const binary = await fetchDescriptionHTMLAndTransform(workspaceSlug, projectId, pageId, cookie);
|
||||
if (binary) {
|
||||
return binary;
|
||||
}
|
||||
}
|
||||
|
||||
return binaryData;
|
||||
};
|
||||
@@ -1,9 +1,8 @@
|
||||
import { Redis } from "ioredis";
|
||||
import { logger } from "@plane/logger";
|
||||
import { getRedisUrl } from "@/core/lib/utils/redis-url";
|
||||
import { ShutdownManager } from "@/core/shutdown-manager";
|
||||
import { shutdownManager } from "@/core/shutdown-manager";
|
||||
|
||||
// Define Redis error interface to handle specific error properties
|
||||
interface RedisError extends Error {
|
||||
code?: string;
|
||||
}
|
||||
@@ -32,9 +31,7 @@ export class RedisManager {
|
||||
const redisUrl = getRedisUrl();
|
||||
|
||||
if (!redisUrl) {
|
||||
logger.warn(
|
||||
"Redis URL is not set, continuing without Redis (you won't be able to sync data between multiple plane live servers)"
|
||||
);
|
||||
shutdownManager.shutdown("Redis URL is not set, shutting down", 1);
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -45,13 +42,12 @@ export class RedisManager {
|
||||
logger.warn(
|
||||
"Initial Redis connection attempt failed. Continuing without Redis (you won't be able to sync data between multiple plane live servers)"
|
||||
);
|
||||
shutdownManager.shutdown("Redis connection failed and could not be recovered", 1);
|
||||
return null;
|
||||
} else {
|
||||
// Once connected at least once, try a few times before giving up
|
||||
if (times > this.maxReconnectAttempts) {
|
||||
logger.error(`Exceeded ${this.maxReconnectAttempts} Redis reconnect attempts. Shutting down the server.`);
|
||||
// Use ShutdownManager to gracefully terminate the server
|
||||
const shutdownManager = ShutdownManager.getInstance();
|
||||
shutdownManager.shutdown("Redis connection lost and could not be recovered", 1);
|
||||
return null; // This will never be reached due to shutdown, but needed for type safety
|
||||
}
|
||||
@@ -76,15 +72,27 @@ export class RedisManager {
|
||||
});
|
||||
|
||||
this.client.on("error", (error: RedisError) => {
|
||||
const fatalErrorCodes = [
|
||||
"ENOTFOUND",
|
||||
"ECONNREFUSED",
|
||||
"ECONNRESET",
|
||||
"ETIMEDOUT",
|
||||
"EHOSTUNREACH",
|
||||
"EPIPE",
|
||||
"WRONGPASS",
|
||||
"NOAUTH",
|
||||
];
|
||||
const fatalMessages = ["WRONGPASS", "NOAUTH", "READONLY", "LOADING", "CLUSTERDOWN", "CONNECTION_BROKEN"];
|
||||
|
||||
if (
|
||||
error?.code === "ENOTFOUND" ||
|
||||
error.message.includes("WRONGPASS") ||
|
||||
error.message.includes("NOAUTH") ||
|
||||
error.message.includes("ECONNREFUSED")
|
||||
(error?.code && fatalErrorCodes.includes(error.code)) ||
|
||||
fatalMessages.some((msg) => error.message.includes(msg))
|
||||
) {
|
||||
if (this.client) this.client.disconnect();
|
||||
shutdownManager.shutdown("Redis connection failed and could not be recovered", 1);
|
||||
} else {
|
||||
logger.warn("Non-fatal Redis error:", error);
|
||||
}
|
||||
logger.warn("Redis error:", error);
|
||||
});
|
||||
|
||||
this.client.on("close", () => {
|
||||
|
||||
@@ -1,18 +1,67 @@
|
||||
// types
|
||||
import { TPage } from "@plane/types";
|
||||
import { TDocumentPayload } from "@plane/types";
|
||||
// services
|
||||
import { API_BASE_URL, APIService } from "@/core/services/api.service";
|
||||
import { env } from "@/env";
|
||||
|
||||
export class PageService extends APIService {
|
||||
// Base params interface
|
||||
export type IBasePageParams<TConfig extends Record<string, any> = Record<string, any>> = {
|
||||
pageId: string;
|
||||
cookie?: string;
|
||||
config: TConfig;
|
||||
};
|
||||
|
||||
export type IFetchDetailsParams<TConfig extends Record<string, any> = Record<string, any>> = IBasePageParams<TConfig>;
|
||||
|
||||
export type IUpdateTitleParams<TConfig extends Record<string, any> = Record<string, any>> = IBasePageParams<TConfig> & {
|
||||
data: { name: string };
|
||||
abortSignal?: AbortSignal;
|
||||
};
|
||||
|
||||
export type IUpdateDescriptionParams<TConfig extends Record<string, any> = Record<string, any>> =
|
||||
IBasePageParams<TConfig> & {
|
||||
data: TDocumentPayload;
|
||||
};
|
||||
|
||||
export abstract class BasePageService extends APIService {
|
||||
constructor() {
|
||||
super(API_BASE_URL);
|
||||
}
|
||||
|
||||
async fetchDetails(workspaceSlug: string, projectId: string, pageId: string, cookie: string): Promise<TPage> {
|
||||
return this.get(`/api/workspaces/${workspaceSlug}/projects/${projectId}/pages/${pageId}/`, {
|
||||
headers: {
|
||||
Cookie: cookie,
|
||||
},
|
||||
/**
|
||||
* Gets the base URL path for API requests based on the context (workspace, project, teamspace)
|
||||
*/
|
||||
protected abstract getBasePath(params: Record<string, any>): string;
|
||||
|
||||
/**
|
||||
* Gets the common headers used for requests
|
||||
*/
|
||||
protected getHeaders(params: { cookie?: string; isBinary?: boolean }): Record<string, string> {
|
||||
const { cookie, isBinary = false } = params;
|
||||
const headers: Record<string, string> = {};
|
||||
|
||||
if (cookie) {
|
||||
headers.Cookie = cookie;
|
||||
}
|
||||
|
||||
if (env.LIVE_SERVER_SECRET_KEY) {
|
||||
headers["live-server-secret-key"] = env.LIVE_SERVER_SECRET_KEY;
|
||||
}
|
||||
|
||||
if (isBinary) {
|
||||
headers["Content-Type"] = "application/octet-stream";
|
||||
}
|
||||
|
||||
return headers;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches page details
|
||||
*/
|
||||
async fetchDetails<TConfig extends Record<string, any>>(params: IFetchDetailsParams<TConfig>) {
|
||||
const { pageId, cookie, config } = params;
|
||||
return this.get(`${this.getBasePath({ pageId, cookie, config })}/`, {
|
||||
headers: this.getHeaders({ cookie }),
|
||||
})
|
||||
.then((response) => response?.data)
|
||||
.catch((error) => {
|
||||
@@ -20,39 +69,93 @@ export class PageService extends APIService {
|
||||
});
|
||||
}
|
||||
|
||||
async fetchDescriptionBinary(workspaceSlug: string, projectId: string, pageId: string, cookie: string): Promise<any> {
|
||||
return this.get(`/api/workspaces/${workspaceSlug}/projects/${projectId}/pages/${pageId}/description/`, {
|
||||
headers: {
|
||||
"Content-Type": "application/octet-stream",
|
||||
Cookie: cookie,
|
||||
},
|
||||
/**
|
||||
* Fetches binary description data for a page
|
||||
*/
|
||||
async fetchDescriptionBinary<TConfig extends Record<string, any>>(params: IFetchDetailsParams<TConfig>) {
|
||||
const { pageId, cookie, config } = params;
|
||||
return this.get(`${this.getBasePath({ pageId, cookie, config })}/description/`, {
|
||||
headers: this.getHeaders({ cookie: cookie, isBinary: true }),
|
||||
responseType: "arraybuffer",
|
||||
})
|
||||
.then((response) => response?.data)
|
||||
.catch((error) => {
|
||||
throw error?.response?.data;
|
||||
console.log("error", error);
|
||||
// throw error?.response?.data;
|
||||
});
|
||||
}
|
||||
|
||||
async updateDescription(
|
||||
workspaceSlug: string,
|
||||
projectId: string,
|
||||
pageId: string,
|
||||
data: {
|
||||
description_binary: string;
|
||||
description_html: string;
|
||||
description: object;
|
||||
},
|
||||
cookie: string
|
||||
): Promise<any> {
|
||||
return this.patch(`/api/workspaces/${workspaceSlug}/projects/${projectId}/pages/${pageId}/description/`, data, {
|
||||
headers: {
|
||||
Cookie: cookie,
|
||||
},
|
||||
/**
|
||||
* Updates the title of a page
|
||||
*/
|
||||
async updateTitle<TConfig extends Record<string, any>>(params: IUpdateTitleParams<TConfig>) {
|
||||
const { config, pageId, data, cookie, abortSignal } = params;
|
||||
|
||||
// Early abort check
|
||||
if (abortSignal?.aborted) {
|
||||
throw new DOMException("Aborted", "AbortError");
|
||||
}
|
||||
|
||||
// Create an abort listener that will reject the pending promise
|
||||
let abortListener: (() => void) | undefined;
|
||||
const abortPromise = new Promise((_, reject) => {
|
||||
if (abortSignal) {
|
||||
abortListener = () => {
|
||||
reject(new DOMException("Aborted", "AbortError"));
|
||||
};
|
||||
abortSignal.addEventListener("abort", abortListener);
|
||||
}
|
||||
});
|
||||
|
||||
try {
|
||||
return await Promise.race([
|
||||
this.patch(`${this.getBasePath({ pageId, cookie, config })}/`, data, {
|
||||
headers: this.getHeaders({ cookie }),
|
||||
signal: abortSignal,
|
||||
})
|
||||
.then((response) => response?.data)
|
||||
.catch((error) => {
|
||||
if (error.name === "AbortError") {
|
||||
throw new DOMException("Aborted", "AbortError");
|
||||
}
|
||||
throw error;
|
||||
}),
|
||||
abortPromise,
|
||||
]);
|
||||
} finally {
|
||||
// Clean up abort listener
|
||||
if (abortSignal && abortListener) {
|
||||
abortSignal.removeEventListener("abort", abortListener);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates the description of a page
|
||||
*/
|
||||
async updateDescription<TConfig extends Record<string, any>>(params: IUpdateDescriptionParams<TConfig>) {
|
||||
const { pageId, data, cookie, config } = params;
|
||||
|
||||
return this.patch(`${this.getBasePath({ pageId, cookie, config })}/description/`, data, {
|
||||
headers: this.getHeaders({ cookie }),
|
||||
})
|
||||
.then((response) => response?.data)
|
||||
.catch((error) => {
|
||||
throw error;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches sub-pages of a page
|
||||
*/
|
||||
async fetchSubPageDetails<TConfig extends Record<string, any>>(params: IFetchDetailsParams<TConfig>) {
|
||||
const { pageId, cookie, config } = params;
|
||||
return this.get(`${this.getBasePath({ pageId, cookie, config })}/sub-pages/`, {
|
||||
headers: this.getHeaders({ cookie: cookie }),
|
||||
})
|
||||
.then((response) => response?.data)
|
||||
.catch((error) => {
|
||||
throw error?.response?.data;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
// services
|
||||
import { ProjectPageConfig } from "@/ce/document-types/project-page-handler";
|
||||
import { BasePageService, IBasePageParams } from "@/core/services/base-page.service";
|
||||
|
||||
/**
|
||||
* Service for handling project page operations
|
||||
*/
|
||||
export class ProjectPageService extends BasePageService {
|
||||
/**
|
||||
* Gets the base URL path for project pages
|
||||
*/
|
||||
protected getBasePath<TConfig extends ProjectPageConfig>(params: IBasePageParams<TConfig>): string {
|
||||
const { pageId, config } = params;
|
||||
const { workspaceSlug, projectId } = config;
|
||||
|
||||
// Handle project pages
|
||||
return `/api/workspaces/${workspaceSlug}/projects/${projectId}/pages/${pageId}`;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
// services
|
||||
import { BasePageService, IBasePageParams } from "@/core/services/base-page.service";
|
||||
|
||||
export interface ServerAgentConfig {}
|
||||
|
||||
export class ServerAgentService extends BasePageService {
|
||||
/**
|
||||
* Gets the base URL path for workspace pages
|
||||
*/
|
||||
protected getBasePath<TConfig extends ServerAgentConfig>(params: IBasePageParams<TConfig>): string {
|
||||
const { pageId } = params;
|
||||
return `/api/pages/${pageId}`;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
// services
|
||||
import { BasePageService, IBasePageParams, IUpdateDescriptionParams } from "@/core/services/base-page.service";
|
||||
|
||||
export interface SyncAgentConfig {}
|
||||
|
||||
export class SyncAgentService extends BasePageService {
|
||||
/**
|
||||
* Gets the base URL path for workspace pages
|
||||
*/
|
||||
protected getBasePath<TConfig extends SyncAgentConfig>(params: IBasePageParams<TConfig>): string {
|
||||
const { pageId } = params;
|
||||
return `/api/pages/${pageId}`;
|
||||
}
|
||||
|
||||
async updateDescription<TConfig extends SyncAgentConfig>(params: IUpdateDescriptionParams<TConfig>) {
|
||||
// no op
|
||||
// since we can't prevent hocuspocus from updating the description after a
|
||||
// sync event, we need to manually override the method to not do anything
|
||||
}
|
||||
}
|
||||
@@ -1,168 +1,74 @@
|
||||
import { Server as HttpServer } from "http";
|
||||
import { Hocuspocus } from "@hocuspocus/server";
|
||||
import { logger } from "@plane/logger";
|
||||
import { RedisManager } from "@/core/lib/redis-manager";
|
||||
// config
|
||||
import { serverConfig } from "@/config/server-config";
|
||||
import { Server } from "http";
|
||||
|
||||
// Global flag to prevent duplicate shutdown sequences
|
||||
let isGlobalShutdownInProgress = false;
|
||||
import { Server as HttpServer } from "http";
|
||||
import { handleError } from "./helpers/error-handling/error-factory";
|
||||
|
||||
/**
|
||||
* ShutdownManager - Handles graceful shutdown of all server components
|
||||
*
|
||||
* Implements the singleton pattern to ensure only one shutdown sequence
|
||||
* can be initiated throughout the application.
|
||||
* Handles graceful shutdown and process signal management for the HTTP server.
|
||||
*/
|
||||
export class ShutdownManager {
|
||||
private static instance: ShutdownManager;
|
||||
class ShutdownManager {
|
||||
private httpServer: HttpServer | null = null;
|
||||
private hocuspocusServer: Hocuspocus | null = null;
|
||||
private isShuttingDown = false;
|
||||
private exitCode = 0;
|
||||
private forceExitTimeout: NodeJS.Timeout | null = null;
|
||||
|
||||
// Private constructor to enforce singleton pattern
|
||||
private constructor() {}
|
||||
|
||||
/**
|
||||
* Get the singleton instance
|
||||
* Register the HTTP server instance to be shut down later.
|
||||
*/
|
||||
public static getInstance(): ShutdownManager {
|
||||
if (!ShutdownManager.instance) {
|
||||
ShutdownManager.instance = new ShutdownManager();
|
||||
}
|
||||
return ShutdownManager.instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* Register server instances that need to be gracefully closed during shutdown
|
||||
*/
|
||||
public register(httpServer: Server, hocuspocusServer: Hocuspocus): void {
|
||||
register({ httpServer }: { httpServer: HttpServer }) {
|
||||
this.httpServer = httpServer;
|
||||
this.hocuspocusServer = hocuspocusServer;
|
||||
logger.info("ShutdownManager registered with server instances");
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a shutdown is in progress
|
||||
* Register process termination signal handlers.
|
||||
*/
|
||||
public isShutdownInProgress(): boolean {
|
||||
return this.isShuttingDown || isGlobalShutdownInProgress;
|
||||
}
|
||||
registerTerminationHandlers(): void {
|
||||
const gracefulTermination = this.getGracefulTerminationHandler();
|
||||
process.on("SIGTERM", gracefulTermination);
|
||||
process.on("SIGINT", gracefulTermination);
|
||||
|
||||
/**
|
||||
* Initiate graceful shutdown sequence
|
||||
* @param reason Reason for shutdown
|
||||
* @param exitCode Process exit code (default: 0)
|
||||
*/
|
||||
public async shutdown(reason: string, exitCode = 0): Promise<void> {
|
||||
// Prevent multiple shutdown attempts
|
||||
if (this.isShuttingDown || isGlobalShutdownInProgress) {
|
||||
logger.warn("Shutdown already in progress, ignoring additional shutdown request");
|
||||
return;
|
||||
}
|
||||
process.on("uncaughtException", (error) => {
|
||||
logger.error("Uncaught exception:", error);
|
||||
|
||||
this.isShuttingDown = true;
|
||||
isGlobalShutdownInProgress = true;
|
||||
this.exitCode = exitCode;
|
||||
handleError(error, {
|
||||
errorType: "internal",
|
||||
component: "process",
|
||||
operation: "uncaughtException",
|
||||
extraContext: { source: "uncaughtException" },
|
||||
});
|
||||
});
|
||||
|
||||
logger.info(`Initiating graceful shutdown: ${reason}`);
|
||||
|
||||
// Create a timeout to force exit if shutdown takes too long
|
||||
this.forceExitTimeout = setTimeout(() => {
|
||||
logger.error("Forcing termination after timeout - some connections may not have closed gracefully.");
|
||||
process.exit(1);
|
||||
}, serverConfig.terminationTimeout || 10000); // Default to 10 seconds if not configured
|
||||
|
||||
try {
|
||||
// Close components in order: Redis, HocusPocus, HTTP server
|
||||
await this.closeRedisConnections();
|
||||
await this.closeHocusPocusServer();
|
||||
await this.closeHttpServer();
|
||||
|
||||
// Wait a bit to allow handles to close
|
||||
await new Promise((resolve) => setTimeout(resolve, 1000));
|
||||
|
||||
console.log("All components shut down successfully");
|
||||
} catch (error) {
|
||||
console.error("Error during graceful shutdown:", error);
|
||||
} finally {
|
||||
// Clear timeout if we've made it this far
|
||||
if (this.forceExitTimeout) {
|
||||
clearTimeout(this.forceExitTimeout);
|
||||
}
|
||||
|
||||
// Give a small delay before exiting to ensure all handles are closed
|
||||
setTimeout(() => {
|
||||
console.info(`Exiting process with code ${this.exitCode}`);
|
||||
process.exit(this.exitCode);
|
||||
}, 100);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close Redis connections
|
||||
*/
|
||||
private async closeRedisConnections(): Promise<void> {
|
||||
console.info("Closing Redis connections...");
|
||||
try {
|
||||
const redisManager = RedisManager.getInstance();
|
||||
const redisClient = redisManager.getClient();
|
||||
|
||||
if (redisClient) {
|
||||
redisClient.disconnect();
|
||||
await redisClient.quit();
|
||||
console.info("Redis connections closed successfully");
|
||||
} else {
|
||||
console.info("No Redis connections to close");
|
||||
}
|
||||
} catch (error) {
|
||||
console.error("Error closing Redis connections:", error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close HocusPocus server
|
||||
*/
|
||||
private async closeHocusPocusServer(): Promise<void> {
|
||||
console.info("Shutting down HocusPocus server...");
|
||||
try {
|
||||
if (this.hocuspocusServer) {
|
||||
await this.hocuspocusServer.destroy();
|
||||
console.info("HocusPocus server shut down successfully");
|
||||
} else {
|
||||
console.info("No HocusPocus server to shut down");
|
||||
}
|
||||
} catch (error) {
|
||||
console.error("Error shutting down HocusPocus server:", error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close HTTP server
|
||||
*/
|
||||
private async closeHttpServer(): Promise<void> {
|
||||
logger.info("Closing HTTP server...");
|
||||
return new Promise<void>((resolve) => {
|
||||
if (!this.httpServer) {
|
||||
console.info("No HTTP server to close");
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
|
||||
// Close all connections
|
||||
this.httpServer.closeAllConnections?.();
|
||||
|
||||
this.httpServer.close((error) => {
|
||||
if (error) {
|
||||
console.error("Error closing HTTP server:", error);
|
||||
} else {
|
||||
console.info("HTTP server closed successfully");
|
||||
}
|
||||
resolve();
|
||||
process.on("unhandledRejection", (reason) => {
|
||||
logger.error("Unhandled rejection:", reason);
|
||||
handleError(reason, {
|
||||
errorType: "internal",
|
||||
component: "process",
|
||||
operation: "unhandledRejection",
|
||||
extraContext: { source: "unhandledRejection" },
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a handler that gracefully shuts down the HTTP server.
|
||||
*/
|
||||
private getGracefulTerminationHandler(): () => Promise<void> {
|
||||
return async () => {
|
||||
logger.info("Signal received, shutting down HTTP server");
|
||||
await this.shutdown();
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Gracefully shuts down the registered HTTP server.
|
||||
*/
|
||||
async shutdown(message?: string, exitCode = 0): Promise<void> {
|
||||
logger.error(`Initiating graceful shutdown ${`${message ? `with message: ${message}` : ""}`}`);
|
||||
|
||||
if (this.httpServer) {
|
||||
this.httpServer.closeAllConnections?.();
|
||||
await new Promise<void>((resolve) => {
|
||||
this.httpServer?.close(() => resolve());
|
||||
});
|
||||
}
|
||||
process.exit(exitCode);
|
||||
}
|
||||
}
|
||||
|
||||
export const shutdownManager = new ShutdownManager();
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
import * as Y from "yjs";
|
||||
|
||||
/**
|
||||
* Recursively finds all XML elements in the tree that match the given criteria
|
||||
*
|
||||
* @param node The root node to start searching from (Y.XmlFragment or Y.XmlElement)
|
||||
* @param nodeName The node name to match (e.g. "pageEmbedComponent")
|
||||
* @param attributeName The attribute name to match
|
||||
* @param attributeValue The attribute value to match or "*" for any value
|
||||
* @returns An array of objects containing the matched node and its path information
|
||||
*/
|
||||
export function findAllElementsRecursive(
|
||||
node: Y.XmlFragment | Y.XmlElement,
|
||||
nodeName: string,
|
||||
attributeName: string,
|
||||
attributeValue: string,
|
||||
path: string[] = []
|
||||
): Array<{
|
||||
node: Y.XmlElement;
|
||||
parent: Y.XmlFragment | Y.XmlElement;
|
||||
indexInParent: number;
|
||||
path: string[];
|
||||
}> {
|
||||
const results: Array<{
|
||||
node: Y.XmlElement;
|
||||
parent: Y.XmlFragment | Y.XmlElement;
|
||||
indexInParent: number;
|
||||
path: string[];
|
||||
}> = [];
|
||||
|
||||
const children = node.toArray();
|
||||
|
||||
for (let i = 0; i < children.length; i++) {
|
||||
const child = children[i];
|
||||
|
||||
if (child instanceof Y.XmlElement) {
|
||||
// Calculate the path to this node
|
||||
const nodePath = [...path];
|
||||
if (child.nodeName) {
|
||||
nodePath.push(`${child.nodeName}[${i}]`);
|
||||
}
|
||||
|
||||
// Check if the current element matches the criteria
|
||||
if (child.nodeName === nodeName) {
|
||||
const attrValue = child.getAttribute(attributeName);
|
||||
// Match if the attribute value is the wildcard "*" or if it matches exactly
|
||||
if (attributeValue === "*" || attrValue === attributeValue) {
|
||||
results.push({
|
||||
node: child,
|
||||
parent: node,
|
||||
indexInParent: i,
|
||||
path: nodePath,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Recursively search in this element's children
|
||||
const nestedResults = findAllElementsRecursive(child, nodeName, attributeName, attributeValue, nodePath);
|
||||
|
||||
results.push(...nestedResults);
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a string representation of the path to a node for logging purposes
|
||||
*
|
||||
* @param pathArray The path array from findAllElementsRecursive
|
||||
* @returns A string representation of the path
|
||||
*/
|
||||
export function getPathString(pathArray: string[]): string {
|
||||
if (pathArray.length === 0) {
|
||||
return "root";
|
||||
}
|
||||
return pathArray.join(" > ");
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert a new node after a specific target node in the XML tree
|
||||
*
|
||||
* @param parent The parent node containing the target node
|
||||
* @param targetIndex The index of the target node in the parent
|
||||
* @param newNode The new node to insert
|
||||
*/
|
||||
export function insertNodeAfter(
|
||||
parent: Y.XmlFragment | Y.XmlElement,
|
||||
targetIndex: number,
|
||||
newNode: Y.XmlElement
|
||||
): void {
|
||||
parent.insert(targetIndex + 1, [newNode]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace a node in the XML tree with a new node
|
||||
*
|
||||
* @param parent The parent node containing the target node
|
||||
* @param targetIndex The index of the target node in the parent
|
||||
* @param newNode The new node to replace the target node with
|
||||
*/
|
||||
export function replaceNode(parent: Y.XmlFragment | Y.XmlElement, targetIndex: number, newNode: Y.XmlElement): void {
|
||||
parent.delete(targetIndex, 1);
|
||||
parent.insert(targetIndex, [newNode]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete a node from the XML tree
|
||||
*
|
||||
* @param parent The parent node containing the target node
|
||||
* @param targetIndex The index of the target node in the parent
|
||||
*/
|
||||
export function deleteNode(parent: Y.XmlFragment | Y.XmlElement, targetIndex: number): void {
|
||||
parent.delete(targetIndex, 1);
|
||||
}
|
||||
@@ -24,6 +24,8 @@ const envSchema = z.object({
|
||||
|
||||
// Graceful termination timeout
|
||||
SHUTDOWN_TIMEOUT: z.string().default("10000").transform(Number),
|
||||
// Live server secret key
|
||||
LIVE_SERVER_SECRET_KEY: z.string(),
|
||||
});
|
||||
|
||||
// Validate the environment variables
|
||||
|
||||
@@ -123,11 +123,16 @@ export const getAllDocumentFormatsFromDocumentEditorBinaryData = (
|
||||
contentBinaryEncoded: string;
|
||||
contentJSON: object;
|
||||
contentHTML: string;
|
||||
titleHTML: string;
|
||||
} => {
|
||||
// encode binary description data
|
||||
const base64Data = convertBinaryDataToBase64String(description);
|
||||
const yDoc = new Y.Doc();
|
||||
Y.applyUpdate(yDoc, description);
|
||||
const title = yDoc.getXmlFragment("title");
|
||||
const titleJSON = yXmlFragmentToProseMirrorRootNode(title, documentEditorSchema).toJSON();
|
||||
// @ts-expect-error tiptap types are incorrect
|
||||
const titleHTML = extractTextFromHTML(generateHTML(titleJSON, DOCUMENT_EDITOR_EXTENSIONS));
|
||||
// convert to JSON
|
||||
const type = yDoc.getXmlFragment("default");
|
||||
const contentJSON = yXmlFragmentToProseMirrorRootNode(type, documentEditorSchema).toJSON();
|
||||
@@ -139,6 +144,7 @@ export const getAllDocumentFormatsFromDocumentEditorBinaryData = (
|
||||
contentBinaryEncoded: base64Data,
|
||||
contentJSON,
|
||||
contentHTML,
|
||||
titleHTML,
|
||||
};
|
||||
};
|
||||
|
||||
@@ -188,3 +194,9 @@ export const convertHTMLDocumentToAllFormats = (args: TConvertHTMLDocumentToAllF
|
||||
|
||||
return allFormats;
|
||||
};
|
||||
|
||||
export const extractTextFromHTML = (html: string): string => {
|
||||
// Use a regex to extract text between tags
|
||||
const textMatch = html.replace(/<[^>]*>/g, "");
|
||||
return textMatch || "";
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user