feat: add proactive delivery HTTP endpoint

plugin-src/host/delivery-http.mjs

@@ -0,0 +1,132 @@
+import { createDeliveryRpcHandler, DELIVERY_ENDPOINTS } from './delivery-rpc.mjs';
+
+export const DELIVERY_HTTP_PATH = '/api/dsh-im/delivery/messages';
+
+const MAX_BODY_BYTES = 1024 * 1024;
+
+const DELIVERY_ERROR_STATUS = Object.freeze({
+  'bad-request': 400,
+  'unknown-bot': 404,
+  'unknown-target': 404,
+  'target-conflict': 409,
+  'invalid-target': 422,
+  'bot-not-connected': 503,
+  'target-rejected': 422,
+  'delivery-failed': 502,
+  cancelled: 408,
+});
+
+class HttpRequestError extends Error {
+  constructor(status, code) {
+    super(code);
+    this.status = status;
+    this.code = code;
+  }
+}
+
+function json(response, status, value, headers = {}) {
+  if (response.destroyed || response.writableEnded) return;
+  const body = JSON.stringify(value);
+  response.writeHead(status, {
+    'cache-control': 'no-store',
+    'content-type': 'application/json; charset=utf-8',
+    'content-length': String(Buffer.byteLength(body)),
+    ...headers,
+  });
+  response.end(body);
+}
+
+function isJsonContentType(value) {
+  return typeof value === 'string'
+    && value.split(';', 1)[0].trim().toLowerCase() === 'application/json';
+}
+
+async function readJsonBody(request) {
+  const declaredLength = Number(request.headers['content-length']);
+  if (Number.isFinite(declaredLength) && declaredLength > MAX_BODY_BYTES) {
+    throw new HttpRequestError(413, 'payload-too-large');
+  }
+
+  const chunks = [];
+  let length = 0;
+  for await (const chunk of request) {
+    const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+    length += buffer.length;
+    if (length > MAX_BODY_BYTES) {
+      throw new HttpRequestError(413, 'payload-too-large');
+    }
+    chunks.push(buffer);
+  }
+
+  try {
+    return JSON.parse(Buffer.concat(chunks, length).toString('utf8'));
+  } catch {
+    throw new HttpRequestError(400, 'bad-request');
+  }
+}
+
+export function createDeliveryHttpHandler(service) {
+  const dispatch = createDeliveryRpcHandler(service);
+  return async (request, response) => {
+    if (request.method !== 'POST') {
+      json(response, 405, {
+        error: { code: 'method-not-allowed', message: 'method-not-allowed', details: {} },
+      }, { allow: 'POST' });
+      return;
+    }
+    if (!isJsonContentType(request.headers['content-type'])) {
+      json(response, 415, {
+        error: { code: 'unsupported-media-type', message: 'unsupported-media-type', details: {} },
+      });
+      return;
+    }
+
+    const abort = new AbortController();
+    const cancel = () => abort.abort();
+    const cancelClosedResponse = () => {
+      if (!response.writableEnded) cancel();
+    };
+    request.once('aborted', cancel);
+    response.once('close', cancelClosedResponse);
+    try {
+      const payload = await readJsonBody(request);
+      const result = await dispatch(DELIVERY_ENDPOINTS.send, payload, abort.signal);
+      if (result.ok) {
+        json(response, 200, result.value);
+        return;
+      }
+      json(response, DELIVERY_ERROR_STATUS[result.error.code] ?? 500, {
+        error: result.error,
+      });
+    } catch (error) {
+      if (error instanceof HttpRequestError) {
+        json(response, error.status, {
+          error: { code: error.code, message: error.code, details: {} },
+        });
+        return;
+      }
+      json(response, 500, {
+        error: { code: 'delivery-failed', message: 'delivery-failed', details: {} },
+      });
+    } finally {
+      request.off('aborted', cancel);
+      response.off('close', cancelClosedResponse);
+    }
+  };
+}
+
+export function installDeliveryHttp(ctx, service) {
+  if (!ctx?.webServer || typeof ctx.webServer.register !== 'function'
+    || typeof ctx.effect !== 'function') {
+    throw new TypeError('DSH Host WebServer is required');
+  }
+  const route = {
+    kind: 'exact',
+    path: DELIVERY_HTTP_PATH,
+    handler: createDeliveryHttpHandler(service),
+  };
+  return ctx.effect(
+    () => ctx.webServer.register(route),
+    `dsh-im: ${DELIVERY_HTTP_PATH}`,
+  );
+}

plugin-src/host/index.mjs

@@ -11,6 +11,7 @@ import { apply as applyWhatsapp } from './channels/whatsapp/index.mjs';
 import { installOutboundArtifactTool } from '../../src/channels/shared/semantic/artifact.mjs';
 import { setImHostLanguage } from '../../src/channels/shared/i18n.mjs';
 import { installDeliveryRpc } from './delivery-rpc.mjs';
+import { installDeliveryHttp } from './delivery-http.mjs';
 import { createDeliveryService } from './delivery-service.mjs';
 import { installUpdateRpc } from './update-rpc.mjs';
 
@@ -32,6 +33,7 @@ function channelConfig(config, name, deliveryService) {
 export function createImHostPlugin(internals = {}) {
   const startUpdate = internals.installUpdateRpc ?? installUpdateRpc;
   const startDelivery = internals.installDeliveryRpc ?? installDeliveryRpc;
+  const startDeliveryHttp = internals.installDeliveryHttp ?? installDeliveryHttp;
   const makeDeliveryService = internals.createDeliveryService ?? createDeliveryService;
   const startFeishu = internals.applyFeishu ?? applyFeishu;
   const startWeixin = internals.applyWeixin ?? applyWeixin;
@@ -77,9 +79,15 @@ export function createImHostPlugin(internals = {}) {
           modern ? ['sessionController', 'workspaceController'] : ['apiProxy'],
           activate,
         );
+        ctx.inject(['webServer'], (httpCtx) => {
+          startDeliveryHttp(httpCtx, deliveryService);
+        });
         return;
       }
       await activate(ctx);
+      if (ctx?.webServer?.register && typeof ctx?.effect === 'function') {
+        startDeliveryHttp(ctx, deliveryService);
+      }
     },
   });