memind/mindspace-service/mindspace-rpc-server.mjs

import http from 'node:http';
import path from 'node:path';
import { fileURLToPath, pathToFileURL } from 'node:url';

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const MCP_TOOL_METHODS = Object.freeze({
  create_dir: 'createDirectory',
  edit_file: 'editFile',
  generate_long_image: 'generateLongImage',
  list_dir: 'listDirectory',
  publish_page: 'publishPage',
  read_file: 'readFile',
  write_binary_file: 'writeBinaryFile',
  write_file: 'writeFile',
});

async function loadContract(env = process.env) {
  const memindRoot = path.resolve(__dirname, env.MINDSPACE_MEMIND_ROOT ?? '../memind-source');
  return import(
    pathToFileURL(path.join(memindRoot, 'mindspace-server-adapter-contract.mjs')).href
  );
}

async function loadMcpScopedTokenVerifier(env = process.env) {
  const memindRoot = path.resolve(__dirname, env.MINDSPACE_MEMIND_ROOT ?? '../memind-source');
  return import(
    pathToFileURL(path.join(memindRoot, 'mindspace-mcp-scoped-token.mjs')).href
  );
}

function parseBearerToken(headers) {
  const auth = String(headers.authorization ?? '');
  if (!auth.startsWith('Bearer ')) return '';
  return auth.slice('Bearer '.length).trim();
}

async function readJsonBody(
  req,
  { maxBytes = 32 * 1024 * 1024 } = {},
) {
  const chunks = [];
  let receivedBytes = 0;
  for await (const chunk of req) {
    const buffer = Buffer.isBuffer(chunk)
      ? chunk
      : Buffer.from(chunk);
    receivedBytes += buffer.length;
    if (receivedBytes > maxBytes) {
      throw Object.assign(
        new Error('MindSpace RPC request body is too large'),
        { code: 'request_body_too_large' },
      );
    }
    chunks.push(buffer);
  }
  const text = Buffer.concat(chunks).toString('utf8').trim();
  if (!text) return {};
  try {
    return JSON.parse(text);
  } catch {
    throw Object.assign(
      new Error('MindSpace RPC request body must be valid JSON'),
      { code: 'invalid_json_body' },
    );
  }
}

function reviveRpcValue(value) {
  if (
    value &&
    typeof value === 'object' &&
    value.type === 'Buffer' &&
    Array.isArray(value.data)
  ) {
    return Buffer.from(value.data);
  }
  if (Array.isArray(value)) return value.map((item) => reviveRpcValue(item));
  if (value && typeof value === 'object') {
    return Object.fromEntries(
      Object.entries(value).map(([key, item]) => [key, reviveRpcValue(item)]),
    );
  }
  return value;
}

function json(res, statusCode, payload) {
  const body = JSON.stringify(payload);
  res.writeHead(statusCode, {
    'Content-Type': 'application/json; charset=utf-8',
    'Content-Length': Buffer.byteLength(body),
  });
  res.end(body);
}

function serializeRpcError(error) {
  return {
    message: error instanceof Error ? error.message : String(error),
    code: error?.code ?? 'internal_error',
    ...(error?.details !== undefined ? { details: error.details } : {}),
  };
}

function resolveRpcErrorStatus(error) {
  switch (error?.code) {
    case 'publication_not_found':
    case 'publication_owner_not_found':
    case 'page_not_found':
    case 'category_not_found':
    case 'static_page_not_found':
      return 404;
    case 'publication_login_required':
      return 401;
    case 'publication_password_required':
    case 'mcp_tool_forbidden':
    case 'mcp_scope_mismatch':
      return 403;
    case 'invalid_mcp_token':
    case 'expired_mcp_token':
      return 401;
    case 'invalid_input':
    case 'invalid_page_input':
    case 'invalid_page_path':
    case 'empty_page_content':
    case 'invalid_publish_input':
    case 'invalid_state_transition':
    case 'slug_conflict':
    case 'security_ack_required':
    case 'invalid_workspace_tool_input':
    case 'invalid_workspace_ref':
    case 'invalid_workspace_path':
    case 'invalid_workspace_package_scope':
    case 'invalid_workspace_user_scope':
    case 'invalid_workspace_page_path':
    case 'invalid_workspace_binary':
    case 'invalid_json_body':
      return 400;
    case 'workspace_entry_not_found':
      return 404;
    case 'workspace_file_too_large':
    case 'request_body_too_large':
      return 413;
    default:
      return 500;
  }
}

export async function createMindSpaceRpcRequestHandler({
  adapter,
  env = process.env,
  logger = console,
  serviceMeta = {},
} = {}) {
  const {
    createMindSpaceServerAdapterContractPayload,
    MINDSPACE_SERVER_ADAPTER_BINDINGS,
    MINDSPACE_SERVER_ADAPTER_CONTRACT_VERSION,
  } = await loadContract(env);
  const operationBasePath = String(
    serviceMeta.operationBasePath ?? env.MINDSPACE_REMOTE_OPERATION_BASE_PATH ?? '/mindspace/v1/adapter',
  )
    .trim()
    .replace(/\/+$/, '') || '/mindspace/v1/adapter';
  const authToken = String(env.MINDSPACE_REMOTE_AUTH_TOKEN ?? '').trim();
  const mcpOperationBasePath = String(
    serviceMeta.mcpOperationBasePath ??
      env.MINDSPACE_MCP_OPERATION_BASE_PATH ??
      '/mindspace/v1/mcp',
  )
    .trim()
    .replace(/\/+$/, '') ||
    '/mindspace/v1/mcp';
  const mcpTokenSecret = String(
    env.MINDSPACE_MCP_TOKEN_SECRET ?? '',
  ).trim();
  if (
    mcpTokenSecret &&
    mcpTokenSecret.length < 16
  ) {
    throw new Error(
      'MINDSPACE_MCP_TOKEN_SECRET must contain at least 16 characters',
    );
  }
  const mcpTokenVerifier = mcpTokenSecret
    ? await loadMcpScopedTokenVerifier(env)
    : null;
  const mcpMaxBodyBytes = Math.max(
    1024,
    Number(
      env.MINDSPACE_MCP_MAX_BODY_BYTES ??
        12 * 1024 * 1024,
    ) || 12 * 1024 * 1024,
  );

  return async function handleMindSpaceRpc(req, res) {
    try {
      const url = new URL(req.url, 'http://127.0.0.1');
      if (req.method === 'GET' && url.pathname === '/health') {
        return json(res, 200, {
          ok: true,
          service: 'mindspace-service',
          adapterKind: adapter.kind,
          implementationStatus: adapter.implementationStatus,
          contractVersion:
            MINDSPACE_SERVER_ADAPTER_CONTRACT_VERSION,
          buildId:
            serviceMeta.buildId ??
            env.MINDSPACE_SERVICE_BUILD_ID ??
            null,
          gitSha:
            serviceMeta.gitSha ??
            env.MINDSPACE_SERVICE_GIT_SHA ??
            null,
          operationBasePath,
          mcpOperationBasePath,
          mcpScopedToolsEnabled:
            Boolean(mcpTokenSecret),
          backgroundJobsManagedLocally: adapter.kind === 'local',
        });
      }
      if (req.method === 'GET' && url.pathname === '/mindspace/v1/contract') {
        return json(
          res,
          200,
          createMindSpaceServerAdapterContractPayload({
            adapter,
            serviceMeta,
            env,
          }),
        );
      }
      if (
        req.method === 'POST' &&
        url.pathname.startsWith(
          `${mcpOperationBasePath}/`,
        )
      ) {
        if (!mcpTokenSecret) {
          return json(res, 503, {
            message:
              'MindSpace MCP scoped tokens are not configured',
          });
        }
        const segments = url.pathname
          .slice(
            mcpOperationBasePath.length + 1,
          )
          .split('/')
          .filter(Boolean);
        if (segments.length !== 1) {
          return json(res, 400, {
            message:
              'Expected /mindspace/v1/mcp/:tool',
          });
        }
        const tool = decodeURIComponent(
          segments[0],
        );
        const method = MCP_TOOL_METHODS[tool];
        if (
          !method ||
          typeof adapter?.workspaceToolService?.[
            method
          ] !== 'function'
        ) {
          return json(res, 404, {
            message: `Unknown MCP tool: ${tool}`,
          });
        }
        const claims =
          mcpTokenVerifier.verifyMindSpaceMcpScopedToken({
            token: parseBearerToken(
              req.headers,
            ),
            secret: mcpTokenSecret,
            tool,
          });
        const body = await readJsonBody(req, {
          maxBytes: mcpMaxBodyBytes,
        });
        const suppliedInput =
          body?.arguments &&
          typeof body.arguments === 'object'
            ? reviveRpcValue(body.arguments)
            : {};
        for (const [
          field,
          expected,
        ] of Object.entries({
          userId: claims.userId,
          sessionId: claims.sessionId,
          packageId: claims.packageId,
          workspaceRef:
            claims.workspaceRef,
        })) {
          const supplied = String(
            suppliedInput?.[field] ?? '',
          ).trim();
          if (
            supplied &&
            supplied !== expected
          ) {
            throw Object.assign(
              new Error(
                `${field} does not match the scoped MCP token`,
              ),
              { code: 'mcp_scope_mismatch' },
            );
          }
        }
        const result =
          await adapter.workspaceToolService[
            method
          ]({
            ...suppliedInput,
            userId: claims.userId,
            sessionId: claims.sessionId,
            packageId: claims.packageId,
            workspaceRef:
              claims.workspaceRef,
          });
        return json(res, 200, {
          ok: true,
          result,
        });
      }
      if (req.method !== 'POST' || !url.pathname.startsWith(`${operationBasePath}/`)) {
        return json(res, 404, { message: 'Not found' });
      }
      if (authToken && parseBearerToken(req.headers) !== authToken) {
        return json(res, 401, { message: 'Unauthorized' });
      }
      const segments = url.pathname.slice(operationBasePath.length + 1).split('/').filter(Boolean);
      if (segments.length !== 2) {
        return json(res, 400, { message: 'Expected /:binding/:method' });
      }
      const [bindingKey, method] = segments.map(decodeURIComponent);
      const service = adapter?.[bindingKey];
      const allowedMethods = MINDSPACE_SERVER_ADAPTER_BINDINGS[bindingKey];
      if (!service || !allowedMethods) {
        return json(res, 404, { message: `Unknown binding: ${bindingKey}` });
      }
      if (!allowedMethods.includes(method)) {
        return json(res, 404, { message: `Unknown method: ${bindingKey}.${method}` });
      }
      const body = await readJsonBody(req);
      const args = Array.isArray(body?.args) ? reviveRpcValue(body.args) : [];
      const result = await service[method](...args);
      return json(res, 200, result);
    } catch (error) {
      const statusCode = resolveRpcErrorStatus(error);
      if (statusCode >= 500) {
        logger.error?.('[MindSpace RPC Error]', error);
      }
      return json(res, statusCode, serializeRpcError(error));
    }
  };
}

export async function startMindSpaceRpcServer({
  adapter,
  env = process.env,
  logger = console,
  serviceMeta = {},
} = {}) {
  const handler = await createMindSpaceRpcRequestHandler({
    adapter,
    env,
    logger,
    serviceMeta,
  });
  const host = env.MINDSPACE_SERVICE_HOST ?? '127.0.0.1';
  const port = Number(env.MINDSPACE_SERVICE_PORT ?? 8082);

  const server = http.createServer((req, res) => {
    void handler(req, res);
  });

  await new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(port, host, () => resolve());
  });

  return {
    server,
    host,
    port,
    close: () =>
      new Promise((resolve, reject) => {
        server.close((error) => {
          if (error) reject(error);
          else resolve();
        });
      }),
  };
}