Add chat gateway

apps/server/src/gateway/Gateway.test.ts

@@ -0,0 +1,153 @@
+import {
+  AdapterName,
+  ChannelId,
+  type GatewayAdapter,
+  GatewayMessageId,
+  GatewayTimestampMillis,
+  IncomingMessage,
+  MessageSender,
+  OutgoingMessage,
+  SentMessage,
+} from "@compass/contracts";
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Stream } from "effect";
+import { makeGateway } from "./Gateway.ts";
+import type { GatewaySettings } from "./GatewayConfig.ts";
+
+const settings = (channels: ReadonlyArray<string>): GatewaySettings => ({
+  allowedChannels: new Set(channels.map((channel) => ChannelId.make(channel))),
+  telegram: {
+    pollingTimeoutSeconds: 30,
+    pollingLimit: 100,
+    retryBaseDelayMillis: 1_000,
+    retryMaxDelayMillis: 30_000,
+    incomingBufferCapacity: 16,
+    maxInboundFileBytes: 50 * 1024 * 1024,
+    deleteWebhook: false,
+    dropPendingUpdates: false,
+  },
+});
+
+const incomingMessage = (channel: string, text: string) =>
+  new IncomingMessage({
+    id: GatewayMessageId.make(text),
+    channel: ChannelId.make(channel),
+    sender: new MessageSender({
+      id: "sender",
+      displayName: "Sender",
+      isBot: false,
+    }),
+    text,
+    files: [],
+    receivedAt: GatewayTimestampMillis.make(1),
+  });
+
+describe("Gateway", () => {
+  it.effect("filters denied channels before materializing their messages", () =>
+    Effect.gen(function* () {
+      let deniedMaterializations = 0;
+      const adapter: GatewayAdapter = {
+        name: AdapterName.make("telegram"),
+        incoming: Stream.fromIterable([
+          {
+            channel: ChannelId.make("telegram:denied"),
+            materialize: Effect.sync(() => {
+              deniedMaterializations += 1;
+              return incomingMessage("telegram:denied", "denied");
+            }),
+          },
+          {
+            channel: ChannelId.make("telegram:allowed"),
+            materialize: Effect.succeed(incomingMessage("telegram:allowed", "allowed")),
+          },
+        ]),
+        send: (message) =>
+          Effect.succeed(
+            new SentMessage({
+              channel: message.channel,
+              messageIds: [GatewayMessageId.make("sent")],
+            }),
+          ),
+      };
+      const gateway = yield* makeGateway([adapter], settings(["telegram:allowed"]));
+      const received = Array.from(yield* gateway.incoming.pipe(Stream.runCollect));
+
+      expect(received.map((message) => message.text)).toEqual(["allowed"]);
+      expect(deniedMaterializations).toBe(0);
+    }),
+  );
+
+  it.effect("authorizes and routes outgoing messages by adapter-qualified channel", () =>
+    Effect.gen(function* () {
+      const sent: Array<OutgoingMessage> = [];
+      const adapter: GatewayAdapter = {
+        name: AdapterName.make("telegram"),
+        incoming: Stream.empty,
+        send: (message) => {
+          sent.push(message);
+          return Effect.succeed(
+            new SentMessage({
+              channel: message.channel,
+              messageIds: [GatewayMessageId.make("42")],
+            }),
+          );
+        },
+      };
+      const gateway = yield* makeGateway([adapter], settings(["telegram:allowed"]));
+      const message = new OutgoingMessage({
+        channel: ChannelId.make("telegram:allowed"),
+        text: "hello",
+        files: [],
+      });
+
+      const result = yield* gateway.send(message);
+      expect(result.messageIds).toEqual(["42"]);
+      expect(sent).toEqual([message]);
+
+      const denied = yield* gateway
+        .send(
+          new OutgoingMessage({
+            channel: ChannelId.make("telegram:denied"),
+            text: "no",
+            files: [],
+          }),
+        )
+        .pipe(Effect.flip);
+      expect(denied._tag).toBe("GatewayAuthorizationError");
+    }),
+  );
+
+  it.effect("consumes an outgoing stream with ordered, bounded delivery", () =>
+    Effect.gen(function* () {
+      const delivered: Array<string> = [];
+      const adapter: GatewayAdapter = {
+        name: AdapterName.make("telegram"),
+        incoming: Stream.empty,
+        send: (message) => {
+          delivered.push(message.text ?? "");
+          return Effect.succeed(
+            new SentMessage({
+              channel: message.channel,
+              messageIds: [GatewayMessageId.make(message.text ?? "sent")],
+            }),
+          );
+        },
+      };
+      const gateway = yield* makeGateway([adapter], settings(["telegram:allowed"]));
+      const outgoing = Stream.fromIterable(
+        ["first", "second"].map(
+          (text) =>
+            new OutgoingMessage({
+              channel: ChannelId.make("telegram:allowed"),
+              text,
+              files: [],
+            }),
+        ),
+      );
+
+      const receipts = Array.from(yield* gateway.sendAll(outgoing).pipe(Stream.runCollect));
+      expect(delivered).toEqual(["first", "second"]);
+      expect(receipts.map((receipt) => receipt.messageIds[0])).toEqual(["first", "second"]);
+    }),
+  );
+});

apps/server/src/gateway/Gateway.ts

@@ -0,0 +1,93 @@
+import {
+  type GatewayAdapter,
+  type GatewayError,
+  GatewayAdapterNotFoundError,
+  GatewayAuthorizationError,
+  GatewayConfigurationError,
+  GatewayInvalidMessageError,
+  type IncomingMessage,
+  type OutgoingMessage,
+  type SentMessage,
+  parseChannelId,
+} from "@compass/contracts";
+import { Context, Effect, Layer, Stream } from "effect";
+import type { GatewaySettings } from "./GatewayConfig.ts";
+import { GatewayConfig } from "./GatewayConfig.ts";
+
+export interface SendAllOptions {
+  readonly concurrency?: number | "unbounded" | undefined;
+}
+
+export interface GatewayService {
+  readonly incoming: Stream.Stream<IncomingMessage, GatewayError>;
+  readonly send: (message: OutgoingMessage) => Effect.Effect<SentMessage, GatewayError>;
+  readonly sendAll: <E, R>(
+    outgoing: Stream.Stream<OutgoingMessage, E, R>,
+    options?: SendAllOptions,
+  ) => Stream.Stream<SentMessage, GatewayError | E, R>;
+}
+
+export class Gateway extends Context.Service<Gateway, GatewayService>()(
+  "@compass/server/gateway/Gateway",
+) {
+  static readonly layer = (adapters: ReadonlyArray<GatewayAdapter>) =>
+    Layer.effect(
+      Gateway,
+      Effect.gen(function* () {
+        const settings = yield* GatewayConfig;
+        return yield* makeGateway(adapters, settings);
+      }),
+    );
+}
+
+export const makeGateway = Effect.fn("Gateway.make")(function* (
+  adapters: ReadonlyArray<GatewayAdapter>,
+  settings: GatewaySettings,
+) {
+  const byName = new Map(adapters.map((adapter) => [adapter.name, adapter] as const));
+  if (byName.size !== adapters.length) {
+    return yield* new GatewayConfigurationError({
+      path: "adapters",
+      message: "Gateway adapter names must be unique",
+      cause: new Error("Duplicate gateway adapter name"),
+    });
+  }
+
+  const incoming = Stream.mergeAll(
+    adapters.map((adapter) => adapter.incoming),
+    { concurrency: "unbounded", bufferSize: 16 },
+  ).pipe(
+    Stream.filter((event) => settings.allowedChannels.has(event.channel)),
+    Stream.mapEffect((event) => event.materialize),
+  );
+
+  const send = Effect.fn("Gateway.send")(function* (message: OutgoingMessage) {
+    if (!settings.allowedChannels.has(message.channel)) {
+      return yield* new GatewayAuthorizationError({ channel: message.channel });
+    }
+    if ((message.text === undefined || message.text.length === 0) && message.files.length === 0) {
+      return yield* new GatewayInvalidMessageError({
+        channel: message.channel,
+        message: "An outgoing message must contain text or at least one file",
+      });
+    }
+    const parsed = parseChannelId(message.channel);
+    const adapter = byName.get(parsed.adapter);
+    if (adapter === undefined) {
+      return yield* new GatewayAdapterNotFoundError({
+        adapter: parsed.adapter,
+        channel: message.channel,
+      });
+    }
+    return yield* adapter.send(message);
+  });
+
+  const sendAll: GatewayService["sendAll"] = (outgoing, options) =>
+    outgoing.pipe(
+      Stream.mapEffect(send, {
+        concurrency: options?.concurrency ?? 1,
+      }),
+    );
+
+  return Gateway.of({ incoming, send, sendAll });
+});

apps/server/src/gateway/GatewayConfig.test.ts

@@ -0,0 +1,54 @@
+import { describe, expect, it } from "@effect/vitest";
+import { NodeServices } from "@effect/platform-node";
+import { Context, Effect, FileSystem, Layer } from "effect";
+import { loadGatewayConfig } from "./GatewayConfig.ts";
+
+describe("GatewayConfig", () => {
+  it.effect("loads strict gateway.json data and applies polling defaults", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const platform = yield* Layer.build(NodeServices.layer);
+        const fileSystem = Context.get(platform, FileSystem.FileSystem);
+        const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "gateway-config-" });
+        const filename = `${directory}/gateway.json`;
+        yield* fileSystem.writeFileString(
+          filename,
+          JSON.stringify({
+            version: 1,
+            allowedChannels: ["telegram:123", "telegram:-100456"],
+            adapters: { telegram: { pollingLimit: 25 } },
+          }),
+        );
+
+        const config = yield* loadGatewayConfig({ filename }).pipe(Effect.provide(platform));
+        expect(Array.from(config.allowedChannels)).toEqual(["telegram:123", "telegram:-100456"]);
+        expect(config.telegram.pollingLimit).toBe(25);
+        expect(config.telegram.pollingTimeoutSeconds).toBe(30);
+        expect(config.telegram.maxInboundFileBytes).toBe(50 * 1024 * 1024);
+        expect(config.telegram.deleteWebhook).toBe(true);
+      }),
+    ),
+  );
+
+  it.effect("rejects unknown keys instead of silently accepting typos", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const platform = yield* Layer.build(NodeServices.layer);
+        const fileSystem = Context.get(platform, FileSystem.FileSystem);
+        const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "gateway-config-" });
+        const filename = `${directory}/gateway.json`;
+        yield* fileSystem.writeFileString(
+          filename,
+          JSON.stringify({ version: 1, allowedChannels: [], allowChannels: ["telegram:1"] }),
+        );
+
+        const error = yield* loadGatewayConfig({ filename }).pipe(
+          Effect.provide(platform),
+          Effect.flip,
+        );
+        expect(error._tag).toBe("GatewayConfigurationError");
+        expect(error.message).toBe("gateway.json is invalid");
+      }),
+    ),
+  );
+});