import * as net from 'net'; import * as plugins from '../congodb.plugins.js'; import { WireProtocol, OP_QUERY } from './WireProtocol.js'; import { CommandRouter } from './CommandRouter.js'; import { MemoryStorageAdapter } from '../storage/MemoryStorageAdapter.js'; import { FileStorageAdapter } from '../storage/FileStorageAdapter.js'; import type { IStorageAdapter } from '../storage/IStorageAdapter.js'; /** * Server configuration options */ export interface ICongoServerOptions { /** Port to listen on (default: 27017) */ port?: number; /** Host to bind to (default: 127.0.0.1) */ host?: string; /** Storage type: 'memory' or 'file' (default: 'memory') */ storage?: 'memory' | 'file'; /** Path for file storage (required if storage is 'file') */ storagePath?: string; /** Enable persistence for memory storage */ persistPath?: string; /** Persistence interval in ms (default: 60000) */ persistIntervalMs?: number; } /** * Connection state for each client */ interface IConnectionState { id: number; socket: net.Socket; buffer: Buffer; authenticated: boolean; database: string; } /** * CongoServer - MongoDB Wire Protocol compatible server * * This server implements the MongoDB wire protocol (OP_MSG) to allow * official MongoDB drivers to connect and perform operations. * * @example * ```typescript * import { CongoServer } from '@push.rocks/smartmongo/congodb'; * import { MongoClient } from 'mongodb'; * * const server = new CongoServer({ port: 27017 }); * await server.start(); * * const client = new MongoClient('mongodb://127.0.0.1:27017'); * await client.connect(); * ``` */ export class CongoServer { private options: Required; private server: net.Server | null = null; private storage: IStorageAdapter; private commandRouter: CommandRouter; private connections: Map = new Map(); private connectionIdCounter = 0; private isRunning = false; private startTime: Date = new Date(); constructor(options: ICongoServerOptions = {}) { this.options = { port: options.port ?? 27017, host: options.host ?? '127.0.0.1', storage: options.storage ?? 'memory', storagePath: options.storagePath ?? './data', persistPath: options.persistPath ?? '', persistIntervalMs: options.persistIntervalMs ?? 60000, }; // Create storage adapter if (this.options.storage === 'file') { this.storage = new FileStorageAdapter(this.options.storagePath); } else { this.storage = new MemoryStorageAdapter({ persistPath: this.options.persistPath || undefined, persistIntervalMs: this.options.persistPath ? this.options.persistIntervalMs : undefined, }); } // Create command router this.commandRouter = new CommandRouter(this.storage, this); } /** * Get the storage adapter (for testing/debugging) */ getStorage(): IStorageAdapter { return this.storage; } /** * Get server uptime in seconds */ getUptime(): number { return Math.floor((Date.now() - this.startTime.getTime()) / 1000); } /** * Get current connection count */ getConnectionCount(): number { return this.connections.size; } /** * Start the server */ async start(): Promise { if (this.isRunning) { throw new Error('Server is already running'); } // Initialize storage await this.storage.initialize(); return new Promise((resolve, reject) => { this.server = net.createServer((socket) => { this.handleConnection(socket); }); this.server.on('error', (err) => { if (!this.isRunning) { reject(err); } else { console.error('Server error:', err); } }); this.server.listen(this.options.port, this.options.host, () => { this.isRunning = true; this.startTime = new Date(); resolve(); }); }); } /** * Stop the server */ async stop(): Promise { if (!this.isRunning || !this.server) { return; } // Close all connections for (const conn of this.connections.values()) { conn.socket.destroy(); } this.connections.clear(); // Close storage await this.storage.close(); return new Promise((resolve) => { this.server!.close(() => { this.isRunning = false; this.server = null; resolve(); }); }); } /** * Handle a new client connection */ private handleConnection(socket: net.Socket): void { const connectionId = ++this.connectionIdCounter; const state: IConnectionState = { id: connectionId, socket, buffer: Buffer.alloc(0), authenticated: true, // No auth required for now database: 'test', }; this.connections.set(connectionId, state); socket.on('data', (data) => { this.handleData(state, Buffer.isBuffer(data) ? data : Buffer.from(data)); }); socket.on('close', () => { this.connections.delete(connectionId); }); socket.on('error', (err) => { // Connection errors are expected when clients disconnect this.connections.delete(connectionId); }); } /** * Handle incoming data from a client */ private handleData(state: IConnectionState, data: Buffer): void { // Append new data to buffer state.buffer = Buffer.concat([state.buffer, data]); // Process messages from buffer this.processMessages(state); } /** * Process complete messages from the buffer */ private async processMessages(state: IConnectionState): Promise { while (state.buffer.length >= 16) { try { const result = WireProtocol.parseMessage(state.buffer); if (!result) { // Not enough data for a complete message break; } const { command, bytesConsumed } = result; // Remove processed bytes from buffer state.buffer = state.buffer.subarray(bytesConsumed); // Process the command const response = await this.commandRouter.route(command); // Encode and send response let responseBuffer: Buffer; if (command.opCode === OP_QUERY) { // Legacy OP_QUERY gets OP_REPLY response responseBuffer = WireProtocol.encodeOpReplyResponse( command.requestID, [response] ); } else { // OP_MSG gets OP_MSG response responseBuffer = WireProtocol.encodeOpMsgResponse( command.requestID, response ); } if (!state.socket.destroyed) { state.socket.write(responseBuffer); } } catch (error: any) { // Send error response const errorResponse = WireProtocol.encodeErrorResponse( 0, // We don't have the requestID at this point 1, error.message || 'Internal error' ); if (!state.socket.destroyed) { state.socket.write(errorResponse); } // Clear buffer on parse errors to avoid infinite loops if (error.message?.includes('opCode') || error.message?.includes('section')) { state.buffer = Buffer.alloc(0); } break; } } } /** * Get the connection URI for this server */ getConnectionUri(): string { return `mongodb://${this.options.host}:${this.options.port}`; } /** * Check if the server is running */ get running(): boolean { return this.isRunning; } /** * Get the port the server is listening on */ get port(): number { return this.options.port; } /** * Get the host the server is bound to */ get host(): string { return this.options.host; } }