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));

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
  );
}

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

async function readJsonBody(req) {
  const chunks = [];
  for await (const chunk of req) {
    chunks.push(chunk);
  }
  const text = Buffer.concat(chunks).toString('utf8').trim();
  return text ? JSON.parse(text) : {};
}

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':
      return 404;
    case 'publication_login_required':
      return 401;
    case 'publication_password_required':
      return 403;
    case 'invalid_input':
    case 'invalid_publish_input':
    case 'invalid_state_transition':
    case 'slug_conflict':
    case 'security_ack_required':
      return 400;
    default:
      return 500;
  }
}

export async function createMindSpaceRpcRequestHandler({
  adapter,
  env = process.env,
  logger = console,
  serviceMeta = {},
} = {}) {
  const {
    MINDSPACE_SERVER_ADAPTER_BINDINGS,
    MINDSPACE_SERVER_ADAPTER_BINDING_KEYS,
  } = 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();

  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,
          operationBasePath,
          backgroundJobsManagedLocally: adapter.kind === 'local',
        });
      }
      if (req.method === 'GET' && url.pathname === '/mindspace/v1/contract') {
        return json(res, 200, {
          bindings: MINDSPACE_SERVER_ADAPTER_BINDINGS,
          bindingKeys: MINDSPACE_SERVER_ADAPTER_BINDING_KEYS,
        });
      }
      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();
        });
      }),
  };
}