diff --git a/.plans/02-chat-gateway.html b/.plans/02-chat-gateway.html new file mode 100644 index 0000000..b9c75e3 --- /dev/null +++ b/.plans/02-chat-gateway.html @@ -0,0 +1,529 @@ + + + + + + Compass Chat Gateway — Implementation Plan + + + + +
+
+
+ 02 + Implementation plan + / + apps/server/src/gateway +
+
+
+

+ Effect-native chat gateway +

+

+ A small, reusable boundary for adapter-qualified channels, finished messages, + local-file attachments, and resilient inbound streams—starting with a complete + Telegram Bot API polling adapter. +

+
+
+
+
Runtime
+
Effect v4
+
+
+
Transport
+
Polling
+
+
+
First adapter
+
Telegram
+
+
+
Message mode
+
Finished only
+
+
+
+
+ +
+
+

Core decision

+

One normalized contract

+

+ Adapters expose an incoming stream and a discrete send effect. The gateway adds + authorization, routing, and a derived outgoing-stream consumer. +

+
+
+

Access model

+

Qualified channel IDs

+

+ telegram:<chat-id> is the + adapter-independent authorization key. Optional topic/thread identity stays separate. +

+
+
+

Files

+

Paths at the boundary

+

+ Outbound attachments reference local paths. Inbound media is downloaded before emission + and points into ~/.compass/files. +

+
+
+ +
+
+
+

01 / Boundary

+

Public API shape

+
+
+
+

+ Gateway.incoming + : Stream<IncomingMessage, GatewayError> +

+

+ Gateway.send(message) + : Effect<SentMessage, GatewayError> +

+

+ Gateway.sendAll(outgoing, options?) + : Stream<SentMessage, GatewayError | E, + R> +

+
+

+ send is the reliable primitive: one + finished message produces one delivery result or one typed failure. + sendAll is the stream-oriented + convenience requested by callers and uses explicit, bounded concurrency. It does not + silently detach fibers or swallow delivery errors. +

+
+
+

Normalized message

+
    +
  • Stable adapter message ID
    • +
  • Qualified channel plus optional thread
    • +
  • Sender identity and received timestamp
    • +
  • Optional text and zero-or-more files
    • +
+
+
+

Attachment metadata

+
    +
  • Absolute local path
    • +
  • Safe display filename
    • +
  • Optional media type and byte size
    • +
  • Kind: image, audio, video, or file
    • +
+
+
+
+
+
+ +
+
+
+

02 / Architecture

+

Layers and ownership

+
+
+
+
+

Application

+

Incoming consumer

+

Outgoing message stream

+
+ +
+

Gateway

+

Authorize + route

+

Merge adapter streams

+
+ +
+

Adapter

+

Telegram polling

+

Bot API + file transfer

+
+
+
+
+

GatewayConfig

+

Schema-decodes gateway.json

+
+
+

GatewayFiles

+

Safe durable file writes

+
+
+

HttpClient

+

Injected Bot API transport

+
+
+

+ The adapter contract contains no Telegram types. Each adapter supplies a stable name, + an incoming stream, and a send implementation. A scoped Telegram layer owns the + long-poll lifecycle, so closing the parent scope interrupts in-flight HTTP and retry + delays cleanly. +

+
+
+
+ +
+
+
+

03 / Configuration

+

+ Non-secret, strict, predictable +

+
+
+ 
{
+  "version": 1,
+  "allowedChannels": [
+    "telegram:123456789",
+    "telegram:-100987654321"
+  ],
+  "adapters": {
+    "telegram": {
+      "pollingTimeoutSeconds": 30,
+      "pollingLimit": 100,
+      "maxInboundFileBytes": 52428800,
+      "deleteWebhook": true,
+      "dropPendingUpdates": false
+    }
+  }
+}
+
+

+ The default path is resolved portably from + HOME or + USERPROFILE. Missing or invalid + configuration fails layer construction with a typed configuration error. +

+

+ The bot token remains outside the JSON file and is read through Effect Config from + TELEGRAM_BOT_TOKEN. An optional API + base URL may be injected for deterministic tests. +

+

+ Authorization is exact-match and deny-by-default. Topics inherit their Telegram + chat’s qualified channel authorization; the optional thread identifier only controls + reply placement. +

+
+
+
+
+ +
+
+
+

04 / Telegram

+

Polling and media flows

+
+
+
+

Incoming

+
    +
  1. + 01Long-poll + getUpdates with the current + offset. +
    2. +
  2. + 02Advance offset monotonically for + every observed update. +
    3. +
  3. + 03Normalize message/channel/sender data + and select the largest photo rendition. +
    4. +
  4. + 04Resolve each attachment using + getFile and enforce its size + limit. +
    5. +
  5. + 05Stream it into an atomic write under + ~/.compass/files without buffering + the full download. +
    6. +
  6. + 06Emit only when the qualified channel + is allowed. +
    7. +
+
+
+

Outgoing

+
    +
  1. + 01Validate the destination and every + local attachment path. +
    2. +
  2. + 02Route by the channel’s adapter + prefix. +
    3. +
  3. + 03Use + sendMessage, + sendPhoto, + sendAudio, + sendVoice, or + sendDocument. +
    4. +
  4. + 04Use text as a caption where + supported; otherwise send text separately. +
    5. +
  5. + 05Return normalized Telegram delivery + identifiers. +
    6. +
+
+
+ Poll failures retry with a capped exponential Effect + Schedule and structured logs. + Successful polls reset backoff. Telegram API envelopes and update payloads are + schema-decoded; malformed individual updates are logged and skipped without killing + the adapter stream. Authentication/configuration failures remain terminal. +
+
+
+
+ +
+
+
+

05 / Files

+

Proposed module tree

+
+
+ 
packages/contracts/src/
+├── gateway.ts                # schemas, domain types, errors, adapter contract
+└── gateway.test.ts
+
+apps/server/src/gateway/
+├── Gateway.ts                 # service, routing, authorization, stream helpers
+├── GatewayConfig.ts           # gateway.json schema + default filesystem layer
+├── GatewayFiles.ts            # inbound persistence + outbound path reads
+├── index.ts                   # public exports
+├── Gateway.test.ts
+├── GatewayConfig.test.ts
+├── GatewayFiles.test.ts
+└── adapters/
+    └── telegram/
+        ├── TelegramAdapter.ts # Effect service/layer + normalization + sending
+        ├── TelegramApi.ts     # typed Bot API HttpClient boundary
+        ├── TelegramModels.ts  # only the Bot API schemas needed here
+        ├── TelegramAdapter.test.ts
+        └── index.ts
+
+
+
+ +
+
+
+

06 / Execution

+

Implementation sequence

+
+
    +
  1. + 01 +
    +

    Model the domain

    +

    + Define Schema-based messages, qualified channel IDs, attachments, delivery + receipts, adapter contract, and tagged error families. +

    +
    +
    2. +
  2. + 02 +
    +

    Build configuration and file services

    +

    + Decode gateway.json, resolve Compass paths, create the files directory, sanitize + filenames, and write downloads atomically. +

    +
    +
    3. +
  3. + 03 +
    +

    Implement the Telegram API boundary

    +

    + Use Effect HttpClient and Schema for getMe, deleteWebhook, getUpdates, getFile, + downloads, JSON requests, and multipart uploads. +

    +
    +
    4. +
  4. + 04 +
    +

    Implement scoped polling

    +

    + Create a lazy incoming stream, track offset, normalize supported messages, persist + attachments, and make interruption immediate and leak-free. +

    +
    +
    5. +
  5. + 05 +
    +

    Compose the gateway

    +

    + Merge adapter streams, enforce allowlisting in both directions, route sends, and + expose bounded outgoing-stream processing. +

    +
    +
    6. +
  6. + 06 +
    +

    Test and publish

    +

    + Exercise config, authorization, ordering, retry, interruption, media + download/upload, then run repository readiness and typecheck gates. +

    +
    +
    7. +
+
+
+ +
+
+
+

07 / Validation

+

Acceptance matrix

+
+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
AreaProofExpected outcome
AuthorizationAllowed and denied channel testsExact-match, deny-by-default
PollingScripted getUpdates responses + Offsets advance; duplicates are avoided +
ResilienceTransient failures + test clockCapped backoff and recovery
ShutdownScoped stream interruptionActive long poll is cancelled
Inbound filesPhoto, audio, voice, document fixturesSafe files exist before emission
Outbound filesMultipart request inspectionCorrect Bot API method and caption
Repository + vp run ready
vp run typecheck + 		Both commands pass
+
+
+
+ +
+
+
+

+ 08 / Deliberate limits +

+

Not in this slice

+
+
+
+ No token/message streaming or edit-in-place responses. +
+
+ No mentions, reactions, commands, buttons, or webhook server. +
+
+ No remote-URL or in-memory outbound attachment source. +
+
+ No history, subscriptions, distributed state, or Chat SDK compatibility layer. +
+
+
+
+ + +
+ + diff --git a/apps/server/src/gateway/Gateway.test.ts b/apps/server/src/gateway/Gateway.test.ts new file mode 100644 index 0000000..61d1c31 --- /dev/null +++ b/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): 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 = []; + 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 = []; + 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"]); + }), + ); +}); diff --git a/apps/server/src/gateway/Gateway.ts b/apps/server/src/gateway/Gateway.ts new file mode 100644 index 0000000..7cf8e07 --- /dev/null +++ b/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; + readonly send: (message: OutgoingMessage) => Effect.Effect; + readonly sendAll: ( + outgoing: Stream.Stream, + options?: SendAllOptions, + ) => Stream.Stream; +} + +export class Gateway extends Context.Service()( + "@compass/server/gateway/Gateway", +) { + static readonly layer = (adapters: ReadonlyArray) => + 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, + 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 }); +}); diff --git a/apps/server/src/gateway/GatewayConfig.test.ts b/apps/server/src/gateway/GatewayConfig.test.ts new file mode 100644 index 0000000..679e1e9 --- /dev/null +++ b/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"); + }), + ), + ); +}); diff --git a/apps/server/src/gateway/GatewayConfig.ts b/apps/server/src/gateway/GatewayConfig.ts new file mode 100644 index 0000000..4ce96dc --- /dev/null +++ b/apps/server/src/gateway/GatewayConfig.ts @@ -0,0 +1,121 @@ +import { ChannelId, GatewayConfigurationError } from "@compass/contracts"; +import { Context, Effect, FileSystem, Layer, Path, Schema } from "effect"; +import { compassDirectory } from "./Paths.ts"; + +const IntegerBetween = (minimum: number, maximum: number) => + Schema.Finite.check( + Schema.isInt(), + Schema.isGreaterThanOrEqualTo(minimum), + Schema.isLessThanOrEqualTo(maximum), + ); +const PositiveInteger = Schema.Finite.check(Schema.isInt(), Schema.isGreaterThan(0)); + +const TelegramSettingsFile = Schema.Struct({ + pollingTimeoutSeconds: Schema.optionalKey(IntegerBetween(0, 50)), + pollingLimit: Schema.optionalKey(IntegerBetween(1, 100)), + retryBaseDelayMillis: Schema.optionalKey(PositiveInteger), + retryMaxDelayMillis: Schema.optionalKey(PositiveInteger), + incomingBufferCapacity: Schema.optionalKey(PositiveInteger), + maxInboundFileBytes: Schema.optionalKey(PositiveInteger), + deleteWebhook: Schema.optionalKey(Schema.Boolean), + dropPendingUpdates: Schema.optionalKey(Schema.Boolean), +}); + +const GatewayConfigFile = Schema.Struct({ + version: Schema.Literal(1), + allowedChannels: Schema.Array(ChannelId), + adapters: Schema.optionalKey( + Schema.Struct({ telegram: Schema.optionalKey(TelegramSettingsFile) }), + ), +}); + +export interface TelegramGatewaySettings { + readonly pollingTimeoutSeconds: number; + readonly pollingLimit: number; + readonly retryBaseDelayMillis: number; + readonly retryMaxDelayMillis: number; + readonly incomingBufferCapacity: number; + readonly maxInboundFileBytes: number; + readonly deleteWebhook: boolean; + readonly dropPendingUpdates: boolean; +} + +export interface GatewaySettings { + readonly allowedChannels: ReadonlySet; + readonly telegram: TelegramGatewaySettings; +} + +export interface GatewayConfigFileOptions { + readonly filename: string; +} + +const defaults: TelegramGatewaySettings = { + pollingTimeoutSeconds: 30, + pollingLimit: 100, + retryBaseDelayMillis: 1_000, + retryMaxDelayMillis: 30_000, + incomingBufferCapacity: 256, + maxInboundFileBytes: 50 * 1024 * 1024, + deleteWebhook: true, + dropPendingUpdates: false, +}; + +export class GatewayConfig extends Context.Service()( + "@compass/server/gateway/GatewayConfig", +) { + static readonly layer = (settings: GatewaySettings) => Layer.succeed(GatewayConfig, settings); +} + +export const loadGatewayConfig = Effect.fn("GatewayConfig.load")(function* ( + options: GatewayConfigFileOptions, +) { + const fileSystem = yield* FileSystem.FileSystem; + const contents = yield* fileSystem.readFileString(options.filename).pipe( + Effect.mapError( + (cause) => + new GatewayConfigurationError({ + path: options.filename, + message: "Could not read the gateway configuration", + cause, + }), + ), + ); + const decoded = yield* Schema.decodeUnknownEffect(Schema.fromJsonString(GatewayConfigFile))( + contents, + { onExcessProperty: "error" }, + ).pipe( + Effect.mapError( + (cause) => + new GatewayConfigurationError({ + path: options.filename, + message: "gateway.json is invalid", + cause, + }), + ), + ); + const telegram = decoded.adapters?.telegram; + return { + allowedChannels: new Set(decoded.allowedChannels), + telegram: { + pollingTimeoutSeconds: telegram?.pollingTimeoutSeconds ?? defaults.pollingTimeoutSeconds, + pollingLimit: telegram?.pollingLimit ?? defaults.pollingLimit, + retryBaseDelayMillis: telegram?.retryBaseDelayMillis ?? defaults.retryBaseDelayMillis, + retryMaxDelayMillis: telegram?.retryMaxDelayMillis ?? defaults.retryMaxDelayMillis, + incomingBufferCapacity: telegram?.incomingBufferCapacity ?? defaults.incomingBufferCapacity, + maxInboundFileBytes: telegram?.maxInboundFileBytes ?? defaults.maxInboundFileBytes, + deleteWebhook: telegram?.deleteWebhook ?? defaults.deleteWebhook, + dropPendingUpdates: telegram?.dropPendingUpdates ?? defaults.dropPendingUpdates, + }, + } satisfies GatewaySettings; +}); + +export const layerFile = (options: GatewayConfigFileOptions) => + Layer.effect(GatewayConfig, loadGatewayConfig(options)); + +export const layerDefault = Layer.unwrap( + Effect.gen(function* () { + const path = yield* Path.Path; + const directory = yield* compassDirectory; + return layerFile({ filename: path.join(directory, "gateway.json") }); + }), +); diff --git a/apps/server/src/gateway/GatewayFiles.test.ts b/apps/server/src/gateway/GatewayFiles.test.ts new file mode 100644 index 0000000..1b7ad65 --- /dev/null +++ b/apps/server/src/gateway/GatewayFiles.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "@effect/vitest"; +import { NodeServices } from "@effect/platform-node"; +import { Context, Effect, FileSystem, Layer, Stream } from "effect"; +import { GatewayFiles, layerAt } from "./GatewayFiles.ts"; + +describe("GatewayFiles", () => { + it.effect("atomically saves sanitized inbound files and reads them back", () => + 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-files-" }); + const context = yield* Layer.build( + layerAt(directory).pipe(Layer.provide(NodeServices.layer)), + ); + const files = Context.get(context, GatewayFiles); + const stored = yield* files.save({ + bytes: Uint8Array.of(1, 2, 3), + preferredName: "../../unsafe name.png", + kind: "image", + mediaType: "image/png", + }); + + expect(stored.name).toBe("unsafe_name.png"); + expect(stored.path.startsWith(directory)).toBe(true); + expect(stored.path.endsWith(".partial")).toBe(false); + expect(Array.from(yield* files.read(stored))).toEqual([1, 2, 3]); + expect((yield* fileSystem.readDirectory(directory)).length).toBe(1); + + const oversized = yield* files + .saveStream({ + stream: Stream.make(Uint8Array.of(1, 2), Uint8Array.of(3, 4)), + preferredName: "too-large.bin", + kind: "file", + maxBytes: 3, + }) + .pipe(Effect.flip); + expect(oversized._tag).toBe("GatewayFileError"); + expect(oversized.operation).toBe("limit incoming file"); + expect((yield* fileSystem.readDirectory(directory)).length).toBe(1); + }), + ), + ); +}); diff --git a/apps/server/src/gateway/GatewayFiles.ts b/apps/server/src/gateway/GatewayFiles.ts new file mode 100644 index 0000000..0a6b33a --- /dev/null +++ b/apps/server/src/gateway/GatewayFiles.ts @@ -0,0 +1,152 @@ +import { GatewayFile, GatewayFileError, type GatewayFileKind } from "@compass/contracts"; +import { Context, Crypto, Effect, FileSystem, Layer, Path, Schema, Stream } from "effect"; +import { compassDirectory } from "./Paths.ts"; + +const UNSAFE_FILENAME_CHARACTERS = /[^a-zA-Z0-9._-]+/g; +const LEADING_DOTS = /^\.+/; +const MAX_FILENAME_LENGTH = 160; + +export interface SaveGatewayFileOptions { + readonly bytes: Uint8Array; + readonly preferredName: string; + readonly kind: GatewayFileKind; + readonly mediaType?: string | undefined; +} + +export interface SaveGatewayFileStreamOptions { + readonly stream: Stream.Stream; + readonly preferredName: string; + readonly kind: GatewayFileKind; + readonly mediaType?: string | undefined; + readonly maxBytes: number; +} + +export interface GatewayFilesService { + readonly directory: string; + readonly save: (options: SaveGatewayFileOptions) => Effect.Effect; + readonly saveStream: ( + options: SaveGatewayFileStreamOptions, + ) => Effect.Effect; + readonly read: (file: GatewayFile) => Effect.Effect; +} + +const sanitizeFilename = (path: Path.Path, input: string): string => { + const basename = path.basename(input).replace(UNSAFE_FILENAME_CHARACTERS, "_"); + const withoutLeadingDots = basename.replace(LEADING_DOTS, ""); + return (withoutLeadingDots || "file").slice(0, MAX_FILENAME_LENGTH); +}; + +export class GatewayFiles extends Context.Service()( + "@compass/server/gateway/GatewayFiles", +) { + static readonly layer = (service: GatewayFilesService) => Layer.succeed(GatewayFiles, service); +} + +const make = Effect.fn("GatewayFiles.make")(function* (directory: string) { + const crypto = yield* Crypto.Crypto; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + yield* fileSystem.makeDirectory(directory, { recursive: true }).pipe( + Effect.mapError( + (cause) => + new GatewayFileError({ + operation: "create directory", + path: directory, + message: "Could not create the Compass gateway files directory", + cause, + }), + ), + ); + + const saveStream = Effect.fn("GatewayFiles.saveStream")(function* ( + options: SaveGatewayFileStreamOptions, + ): Effect.fn.Return { + const identifier = yield* crypto.randomUUIDv7.pipe( + Effect.mapError( + (cause) => + new GatewayFileError({ + operation: "generate file name", + path: directory, + message: "Could not generate a unique attachment name", + cause, + }), + ), + ); + const filename = `${identifier}-${sanitizeFilename(path, options.preferredName)}`; + const destination = path.join(directory, filename); + const temporary = `${destination}.partial`; + let size = 0; + yield* options.stream.pipe( + Stream.mapEffect((chunk) => + Effect.suspend(() => { + size += chunk.byteLength; + return size <= options.maxBytes + ? Effect.succeed(chunk) + : Effect.fail( + new GatewayFileError({ + operation: "limit incoming file", + path: destination, + message: `Incoming attachment exceeds the ${options.maxBytes} byte limit`, + cause: new Error("Incoming gateway attachment is too large"), + }), + ); + }), + ), + Stream.run(fileSystem.sink(temporary)), + Effect.andThen(fileSystem.rename(temporary, destination)), + Effect.onError(() => fileSystem.remove(temporary, { force: true }).pipe(Effect.ignore)), + Effect.mapError((cause) => + Schema.is(GatewayFileError)(cause) + ? cause + : new GatewayFileError({ + operation: "save incoming file", + path: destination, + message: "Could not persist an incoming gateway attachment", + cause, + }), + ), + ); + return new GatewayFile({ + path: destination, + name: sanitizeFilename(path, options.preferredName), + kind: options.kind, + ...(options.mediaType === undefined ? {} : { mediaType: options.mediaType }), + size, + }); + }); + + const save = (options: SaveGatewayFileOptions) => + saveStream({ + stream: Stream.succeed(options.bytes), + preferredName: options.preferredName, + kind: options.kind, + mediaType: options.mediaType, + maxBytes: options.bytes.byteLength, + }); + + const read = Effect.fn("GatewayFiles.read")(function* (file: GatewayFile) { + return yield* fileSystem.readFile(file.path).pipe( + Effect.mapError( + (cause) => + new GatewayFileError({ + operation: "read outgoing file", + path: file.path, + message: `Could not read outgoing attachment ${file.name}`, + cause, + }), + ), + ); + }); + + return GatewayFiles.of({ directory, save, saveStream, read }); +}); + +export const layerAt = (directory: string) => Layer.effect(GatewayFiles, make(directory)); + +export const layerDefault = Layer.unwrap( + Effect.gen(function* () { + const path = yield* Path.Path; + const directory = yield* compassDirectory; + return layerAt(path.join(directory, "files")); + }), +); diff --git a/apps/server/src/gateway/Paths.ts b/apps/server/src/gateway/Paths.ts new file mode 100644 index 0000000..811f6d8 --- /dev/null +++ b/apps/server/src/gateway/Paths.ts @@ -0,0 +1,7 @@ +import { Config, Effect, Path } from "effect"; + +export const compassDirectory = Effect.gen(function* () { + const path = yield* Path.Path; + const home = yield* Config.string("HOME").pipe(Config.orElse(() => Config.string("USERPROFILE"))); + return path.join(home, ".compass"); +}); diff --git a/apps/server/src/gateway/adapters/telegram/TelegramAdapter.test.ts b/apps/server/src/gateway/adapters/telegram/TelegramAdapter.test.ts new file mode 100644 index 0000000..dcacfcd --- /dev/null +++ b/apps/server/src/gateway/adapters/telegram/TelegramAdapter.test.ts @@ -0,0 +1,262 @@ +import { + ChannelId, + GatewayFile, + type GatewayAdapterError, + GatewayFileError, + GatewayMessageId, + OutgoingMessage, +} from "@compass/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import { Context, Effect, Latch, Layer, Stream } from "effect"; +import { makeGateway } from "../../Gateway.ts"; +import { GatewayConfig, type GatewaySettings } from "../../GatewayConfig.ts"; +import { GatewayFiles, type GatewayFilesService } from "../../GatewayFiles.ts"; +import { TelegramAdapter } from "./TelegramAdapter.ts"; +import { TelegramApi, type TelegramApiService } from "./TelegramApi.ts"; + +const settings = (allowedChannels: ReadonlyArray): GatewaySettings => ({ + allowedChannels: new Set(allowedChannels.map((channel) => ChannelId.make(channel))), + telegram: { + pollingTimeoutSeconds: 30, + pollingLimit: 100, + retryBaseDelayMillis: 1, + retryMaxDelayMillis: 10, + incomingBufferCapacity: 16, + maxInboundFileBytes: 50 * 1024 * 1024, + deleteWebhook: true, + dropPendingUpdates: false, + }, +}); + +const baseApi = (): TelegramApiService => ({ + getMe: Effect.succeed({ id: 99, is_bot: true, first_name: "Compass", username: "compass_bot" }), + deleteWebhook: () => Effect.void, + getUpdates: () => Effect.never, + getFile: () => Effect.die("getFile was not expected"), + download: () => Stream.die("download was not expected"), + sendText: () => Effect.die("sendText was not expected"), + sendMedia: () => Effect.die("sendMedia was not expected"), +}); + +const baseFiles = (): GatewayFilesService => ({ + directory: "C:/gateway-files", + save: () => Effect.die("save was not expected"), + saveStream: () => Effect.die("saveStream was not expected"), + read: () => Effect.die("read was not expected"), +}); + +const adapterLayer = ( + config: GatewaySettings, + api: TelegramApiService, + files: GatewayFilesService, +): Layer.Layer => + TelegramAdapter.layer.pipe( + Layer.provide( + Layer.mergeAll( + GatewayConfig.layer(config), + Layer.succeed(TelegramApi, api), + GatewayFiles.layer(files), + ), + ), + ); + +describe("TelegramAdapter", () => { + it.effect("polls with offsets and lets the gateway reject media before download", () => + Effect.scoped( + Effect.gen(function* () { + const secondPoll = yield* Latch.make(); + const pollingRequests: Array<{ readonly offset?: number | undefined }> = []; + let deleteWebhookCalls = 0; + const api: TelegramApiService = { + ...baseApi(), + deleteWebhook: () => { + deleteWebhookCalls += 1; + return Effect.void; + }, + getUpdates: (request) => { + pollingRequests.push(request); + if (pollingRequests.length === 1) { + return Effect.succeed([ + { + update_id: 10, + message: { + message_id: 100, + date: 1, + chat: { id: 1, type: "private" }, + from: { id: 1, is_bot: false, first_name: "Denied" }, + photo: [{ file_id: "secret-photo", width: 10, height: 10 }], + }, + }, + { + update_id: 11, + message: { + message_id: 101, + date: 2, + chat: { id: 2, type: "private" }, + from: { id: 2, is_bot: false, first_name: "Allowed" }, + text: "hello", + }, + }, + ]); + } + return secondPoll.open.pipe(Effect.andThen(Effect.never)); + }, + }; + const config = settings(["telegram:2"]); + const context = yield* Layer.build(adapterLayer(config, api, baseFiles())); + const adapter = Context.get(context, TelegramAdapter); + const gateway = yield* makeGateway([adapter], config); + + const received = Array.from( + yield* gateway.incoming.pipe(Stream.take(1), Stream.runCollect), + ); + yield* secondPoll.await; + + expect(received.map((message) => message.text)).toEqual(["hello"]); + expect(pollingRequests[1]?.offset).toBe(12); + expect(deleteWebhookCalls).toBe(1); + }), + ), + ); + + it.effect("uploads finished image and voice attachments from local paths", () => + Effect.scoped( + Effect.gen(function* () { + const mediaRequests: Array<{ + readonly method: string; + readonly caption?: string | undefined; + readonly filename: string; + }> = []; + let nextMessageId = 40; + const api: TelegramApiService = { + ...baseApi(), + sendMedia: (request) => { + mediaRequests.push(request); + nextMessageId += 1; + return Effect.succeed({ + message_id: nextMessageId, + date: 1, + chat: { id: 2, type: "private" }, + }); + }, + }; + const files: GatewayFilesService = { + ...baseFiles(), + read: () => Effect.succeed(Uint8Array.of(1, 2, 3)), + }; + const config = { + ...settings(["telegram:2"]), + telegram: { ...settings([]).telegram, deleteWebhook: false }, + }; + const context = yield* Layer.build(adapterLayer(config, api, files)); + const adapter = Context.get(context, TelegramAdapter); + const result = yield* adapter.send( + new OutgoingMessage({ + channel: ChannelId.make("telegram:2"), + text: "caption", + files: [ + new GatewayFile({ + path: "C:/tmp/image.png", + name: "image.png", + kind: "image", + mediaType: "image/png", + }), + new GatewayFile({ + path: "C:/tmp/voice.ogg", + name: "voice.ogg", + kind: "audio", + mediaType: "audio/ogg", + }), + ], + }), + ); + + expect( + mediaRequests.map(({ method, caption, filename }) => ({ method, caption, filename })), + ).toEqual([ + { method: "sendPhoto", caption: "caption", filename: "image.png" }, + { method: "sendVoice", caption: undefined, filename: "voice.ogg" }, + ]); + expect(result.messageIds).toEqual([ + GatewayMessageId.make("41"), + GatewayMessageId.make("42"), + ]); + }), + ), + ); + + it.effect("downloads and materializes supported files for allowed incoming channels", () => + Effect.scoped( + Effect.gen(function* () { + let polled = false; + const savedKinds: Array = []; + const api: TelegramApiService = { + ...baseApi(), + getUpdates: () => { + if (polled) return Effect.never; + polled = true; + return Effect.succeed([ + { + update_id: 20, + message: { + message_id: 200, + date: 3, + chat: { id: 2, type: "private" }, + from: { id: 2, is_bot: false, first_name: "Allowed" }, + caption: "two files", + photo: [{ file_id: "photo", width: 20, height: 20 }], + voice: { file_id: "voice", mime_type: "audio/ogg" }, + }, + }, + ]); + }, + getFile: (fileId) => + Effect.succeed({ file_id: fileId, file_path: `${fileId}.bin`, file_size: 3 }), + download: () => Stream.succeed(Uint8Array.of(1, 2, 3)), + }; + const files: GatewayFilesService = { + ...baseFiles(), + saveStream: (options) => { + savedKinds.push(options.kind); + return options.stream.pipe( + Stream.runDrain, + Effect.mapError( + (cause) => + new GatewayFileError({ + operation: "test save", + path: "C:/gateway-files", + message: "Test attachment stream failed", + cause, + }), + ), + Effect.as( + new GatewayFile({ + path: `C:/gateway-files/${options.preferredName}`, + name: options.preferredName, + kind: options.kind, + ...(options.mediaType === undefined ? {} : { mediaType: options.mediaType }), + size: 3, + }), + ), + ); + }, + }; + const config = { + ...settings(["telegram:2"]), + telegram: { ...settings([]).telegram, deleteWebhook: false }, + }; + const context = yield* Layer.build(adapterLayer(config, api, files)); + const adapter = Context.get(context, TelegramAdapter); + const gateway = yield* makeGateway([adapter], config); + + const message = Array.from( + yield* gateway.incoming.pipe(Stream.take(1), Stream.runCollect), + )[0]; + + expect(message?.text).toBe("two files"); + expect(message?.files.map((file) => file.kind)).toEqual(["image", "audio"]); + expect(savedKinds).toEqual(["image", "audio"]); + }), + ), + ); +}); diff --git a/apps/server/src/gateway/adapters/telegram/TelegramAdapter.ts b/apps/server/src/gateway/adapters/telegram/TelegramAdapter.ts new file mode 100644 index 0000000..2a056f7 --- /dev/null +++ b/apps/server/src/gateway/adapters/telegram/TelegramAdapter.ts @@ -0,0 +1,353 @@ +import { + AdapterName, + ChannelId, + type GatewayAdapter, + type GatewayFile, + GatewayFileError, + type GatewayFileKind, + GatewayMessageId, + GatewayProtocolError, + GatewayTimestampMillis, + type IncomingAdapterEvent, + IncomingMessage, + MessageSender, + type OutgoingMessage, + SentMessage, + ThreadId, + parseChannelId, +} from "@compass/contracts"; +import { Context, Effect, Layer, Queue, Schedule, Schema, Stream } from "effect"; +import { GatewayConfig } from "../../GatewayConfig.ts"; +import { GatewayFiles } from "../../GatewayFiles.ts"; +import { TelegramApi, type TelegramMediaRequest } from "./TelegramApi.ts"; +import { + type TelegramMessage, + type TelegramPhotoSize, + TelegramUpdate, + TelegramUpdateIdentity, +} from "./TelegramModels.ts"; + +const TELEGRAM_ADAPTER = AdapterName.make("telegram"); +const EMPTY_POLL_DELAY = "25 millis"; + +interface TelegramAttachment { + readonly fileId: string; + readonly preferredName: string; + readonly kind: GatewayFileKind; + readonly mediaType?: string | undefined; +} + +const bestPhoto = ( + photos: ReadonlyArray | undefined, +): TelegramPhotoSize | undefined => { + let selected: TelegramPhotoSize | undefined; + let area = -1; + for (const photo of photos ?? []) { + const candidateArea = photo.width * photo.height; + if (candidateArea > area) { + selected = photo; + area = candidateArea; + } + } + return selected; +}; + +const attachmentsOf = (message: TelegramMessage): ReadonlyArray => { + const attachments: Array = []; + const photo = bestPhoto(message.photo); + if (photo !== undefined) { + attachments.push({ + fileId: photo.file_id, + preferredName: `photo-${message.message_id}.jpg`, + kind: "image", + mediaType: "image/jpeg", + }); + } + if (message.audio !== undefined) { + attachments.push({ + fileId: message.audio.file_id, + preferredName: message.audio.file_name ?? `audio-${message.message_id}`, + kind: "audio", + mediaType: message.audio.mime_type, + }); + } + if (message.voice !== undefined) { + attachments.push({ + fileId: message.voice.file_id, + preferredName: `voice-${message.message_id}.ogg`, + kind: "audio", + mediaType: message.voice.mime_type ?? "audio/ogg", + }); + } + if (message.video !== undefined) { + attachments.push({ + fileId: message.video.file_id, + preferredName: message.video.file_name ?? `video-${message.message_id}.mp4`, + kind: "video", + mediaType: message.video.mime_type ?? "video/mp4", + }); + } + if (message.video_note !== undefined) { + attachments.push({ + fileId: message.video_note.file_id, + preferredName: `video-note-${message.message_id}.mp4`, + kind: "video", + mediaType: message.video_note.mime_type ?? "video/mp4", + }); + } + if (message.document !== undefined) { + attachments.push({ + fileId: message.document.file_id, + preferredName: message.document.file_name ?? `document-${message.message_id}`, + kind: "file", + mediaType: message.document.mime_type, + }); + } + return attachments; +}; + +const hasSupportedContent = (message: TelegramMessage): boolean => + message.text !== undefined || message.caption !== undefined || attachmentsOf(message).length > 0; + +const senderOf = (message: TelegramMessage): MessageSender => { + if (message.from !== undefined) { + const displayName = [message.from.first_name, message.from.last_name] + .filter((part) => part !== undefined && part.length > 0) + .join(" "); + return new MessageSender({ + id: String(message.from.id), + displayName: displayName || message.from.username || String(message.from.id), + ...(message.from.username === undefined ? {} : { username: message.from.username }), + isBot: message.from.is_bot, + }); + } + const senderChat = message.sender_chat ?? message.chat; + return new MessageSender({ + id: String(senderChat.id), + displayName: + senderChat.title ?? senderChat.username ?? senderChat.first_name ?? String(senderChat.id), + ...(senderChat.username === undefined ? {} : { username: senderChat.username }), + isBot: false, + }); +}; + +const inferMediaType = (file: GatewayFile): string | undefined => { + if (file.mediaType !== undefined) { + return file.mediaType; + } + const lower = file.name.toLowerCase(); + if (lower.endsWith(".jpg") || lower.endsWith(".jpeg")) return "image/jpeg"; + if (lower.endsWith(".png")) return "image/png"; + if (lower.endsWith(".gif")) return "image/gif"; + if (lower.endsWith(".mp3")) return "audio/mpeg"; + if (lower.endsWith(".ogg") || lower.endsWith(".opus")) return "audio/ogg"; + if (lower.endsWith(".mp4")) return "video/mp4"; + return undefined; +}; + +const mediaTarget = (file: GatewayFile): Pick => { + const mediaType = inferMediaType(file); + if (file.kind === "image") return { method: "sendPhoto", field: "photo" }; + if (file.kind === "video") return { method: "sendVideo", field: "video" }; + if (file.kind === "audio" && mediaType === "audio/ogg") { + return { method: "sendVoice", field: "voice" }; + } + if (file.kind === "audio") return { method: "sendAudio", field: "audio" }; + return { method: "sendDocument", field: "document" }; +}; + +const parseThread = ( + message: OutgoingMessage, +): Effect.Effect => + Effect.suspend(() => { + if (message.thread === undefined) { + return Effect.as(Effect.void, undefined as number | undefined); + } + const value = Number(message.thread); + return Number.isSafeInteger(value) && value > 0 + ? Effect.succeed(value) + : Effect.fail( + new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation: "parse thread", + message: `Telegram thread ${message.thread} is not a positive integer`, + cause: new Error("Invalid Telegram message_thread_id"), + }), + ); + }); + +export class TelegramAdapter extends Context.Service()( + "@compass/server/gateway/adapters/telegram/TelegramAdapter", +) { + static readonly layer = Layer.effect( + TelegramAdapter, + Effect.suspend(() => makeTelegramAdapter), + ); +} + +export const makeTelegramAdapter = Effect.gen(function* () { + const api = yield* TelegramApi; + const files = yield* GatewayFiles; + const config = yield* GatewayConfig; + const settings = config.telegram; + const me = yield* api.getMe; + yield* Effect.logInfo("Telegram gateway adapter authenticated", { + botUserId: me.id, + username: me.username, + }); + if (settings.deleteWebhook) { + yield* api.deleteWebhook(settings.dropPendingUpdates); + } + + const queue = yield* Queue.bounded(settings.incomingBufferCapacity); + yield* Effect.addFinalizer(() => Queue.shutdown(queue)); + + const materialize = Effect.fn("TelegramAdapter.materialize")(function* ( + message: TelegramMessage, + channel: ChannelId, + ) { + const saved = yield* Effect.forEach( + attachmentsOf(message), + Effect.fn(function* (attachment) { + const telegramFile = yield* api.getFile(attachment.fileId); + if (telegramFile.file_path === undefined) { + return yield* new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation: "getFile", + message: `Telegram did not return a path for file ${attachment.fileId}`, + cause: new Error("Telegram file_path is missing"), + }); + } + if ( + telegramFile.file_size !== undefined && + telegramFile.file_size > settings.maxInboundFileBytes + ) { + return yield* new GatewayFileError({ + operation: "limit incoming file", + path: telegramFile.file_path, + message: `Incoming attachment exceeds the ${settings.maxInboundFileBytes} byte limit`, + cause: new Error("Telegram attachment is too large"), + }); + } + return yield* files.saveStream({ + stream: api.download(telegramFile.file_path), + preferredName: attachment.preferredName, + kind: attachment.kind, + mediaType: attachment.mediaType, + maxBytes: settings.maxInboundFileBytes, + }); + }), + { concurrency: 2 }, + ); + return new IncomingMessage({ + id: GatewayMessageId.make(String(message.message_id)), + channel, + ...(message.message_thread_id === undefined + ? {} + : { thread: ThreadId.make(String(message.message_thread_id)) }), + sender: senderOf(message), + ...(message.text === undefined && message.caption === undefined + ? {} + : { text: message.text ?? message.caption }), + files: saved, + receivedAt: GatewayTimestampMillis.make(Math.max(0, Math.trunc(message.date * 1_000))), + }); + }); + + const processUpdate = Effect.fn("TelegramAdapter.processUpdate")(function* (input: unknown) { + const identity = yield* Schema.decodeUnknownEffect(TelegramUpdateIdentity)(input).pipe( + Effect.option, + ); + if (identity._tag === "None") { + yield* Effect.logWarning("Ignoring Telegram update without a valid update_id"); + return undefined; + } + const decoded = yield* Schema.decodeUnknownEffect(TelegramUpdate)(input).pipe(Effect.option); + if (decoded._tag === "None") { + yield* Effect.logWarning("Ignoring malformed Telegram update", { + updateId: identity.value.update_id, + }); + return identity.value.update_id; + } + const message = decoded.value.message ?? decoded.value.channel_post; + if (message === undefined || !hasSupportedContent(message)) { + return identity.value.update_id; + } + const channel = ChannelId.make(`telegram:${message.chat.id}`); + yield* Queue.offer(queue, { + channel, + materialize: materialize(message, channel), + }); + return identity.value.update_id; + }); + + const retrySchedule = Schedule.min([ + Schedule.exponential(settings.retryBaseDelayMillis), + Schedule.spaced(Math.max(settings.retryBaseDelayMillis, settings.retryMaxDelayMillis)), + ]).pipe( + Schedule.tap(({ input }) => + Effect.logWarning("Telegram polling failed; retrying", { error: input }), + ), + ); + + const polling = Effect.gen(function* () { + let offset: number | undefined; + while (true) { + const updates = yield* api + .getUpdates({ + offset, + timeout: settings.pollingTimeoutSeconds, + limit: settings.pollingLimit, + }) + .pipe(Effect.retry(retrySchedule)); + for (const update of updates) { + const updateId = yield* processUpdate(update); + if (updateId !== undefined) { + offset = Math.max(offset ?? 0, updateId + 1); + } + } + if (updates.length === 0) { + yield* Effect.sleep(EMPTY_POLL_DELAY); + } + } + }); + yield* polling.pipe(Effect.forkScoped({ startImmediately: true })); + + const send = Effect.fn("TelegramAdapter.send")(function* (message: OutgoingMessage) { + const { nativeChannel } = parseChannelId(message.channel); + const threadId = yield* parseThread(message); + const messageIds: Array = []; + if (message.files.length === 0) { + const sent = yield* api.sendText(nativeChannel, threadId, message.text ?? ""); + messageIds.push(GatewayMessageId.make(String(sent.message_id))); + } else { + for (let index = 0; index < message.files.length; index += 1) { + const file = message.files[index]; + if (file === undefined) continue; + const bytes = yield* files.read(file); + const target = mediaTarget(file); + const sent = yield* api.sendMedia({ + ...target, + chatId: nativeChannel, + threadId, + filename: file.name, + mediaType: inferMediaType(file), + bytes, + caption: index === 0 ? message.text : undefined, + }); + messageIds.push(GatewayMessageId.make(String(sent.message_id))); + } + } + return new SentMessage({ + channel: message.channel, + ...(message.thread === undefined ? {} : { thread: message.thread }), + messageIds, + }); + }); + + return TelegramAdapter.of({ + name: TELEGRAM_ADAPTER, + incoming: Stream.fromQueue(queue), + send, + }); +}); diff --git a/apps/server/src/gateway/adapters/telegram/TelegramApi.test.ts b/apps/server/src/gateway/adapters/telegram/TelegramApi.test.ts new file mode 100644 index 0000000..b5e5620 --- /dev/null +++ b/apps/server/src/gateway/adapters/telegram/TelegramApi.test.ts @@ -0,0 +1,81 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Redacted } from "effect"; +import { HttpClient, HttpClientResponse } from "effect/unstable/http"; +import { makeTelegramApi } from "./TelegramApi.ts"; + +describe("TelegramApi", () => { + it.effect("sends local media as multipart form data through Effect HttpClient", () => + Effect.gen(function* () { + let requestedUrl = ""; + let requestedForm: FormData | undefined; + const client = HttpClient.make((request) => { + requestedUrl = request.url; + if (request.body._tag === "FormData") { + requestedForm = request.body.formData; + } + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response( + JSON.stringify({ + ok: true, + result: { + message_id: 7, + date: 1, + chat: { id: 123, type: "private" }, + }, + }), + { headers: { "content-type": "application/json" } }, + ), + ), + ); + }); + const api = yield* makeTelegramApi({ + botToken: Redacted.make("secret-token"), + apiBaseUrl: "https://telegram.invalid/", + }).pipe(Effect.provideService(HttpClient.HttpClient, client)); + + const response = yield* api.sendMedia({ + method: "sendPhoto", + field: "photo", + chatId: "123", + threadId: 9, + filename: "image.png", + mediaType: "image/png", + bytes: Uint8Array.of(1, 2, 3), + caption: "caption", + }); + + expect(response.message_id).toBe(7); + expect(requestedUrl).toBe("https://telegram.invalid/botsecret-token/sendPhoto"); + expect(requestedForm?.get("chat_id")).toBe("123"); + expect(requestedForm?.get("message_thread_id")).toBe("9"); + expect(requestedForm?.get("caption")).toBe("caption"); + expect(requestedForm?.get("photo")).toBeInstanceOf(File); + }), + ); + + it.effect("maps unsuccessful Bot API envelopes to typed protocol errors", () => + Effect.gen(function* () { + const client = HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response( + JSON.stringify({ ok: false, error_code: 401, description: "Unauthorized" }), + ), + ), + ), + ); + const api = yield* makeTelegramApi({ + botToken: Redacted.make("bad-token"), + apiBaseUrl: "https://telegram.invalid", + }).pipe(Effect.provideService(HttpClient.HttpClient, client)); + + const error = yield* api.getMe.pipe(Effect.flip); + expect(error._tag).toBe("GatewayProtocolError"); + expect(error.operation).toBe("getMe"); + expect(error.message).toBe("Unauthorized"); + }), + ); +}); diff --git a/apps/server/src/gateway/adapters/telegram/TelegramApi.ts b/apps/server/src/gateway/adapters/telegram/TelegramApi.ts new file mode 100644 index 0000000..c022e2f --- /dev/null +++ b/apps/server/src/gateway/adapters/telegram/TelegramApi.ts @@ -0,0 +1,288 @@ +import { + AdapterName, + GatewayProtocolError, + GatewayTransportError, + type GatewayAdapterError, +} from "@compass/contracts"; +import { Config, Context, Effect, Layer, Redacted, Schema, Stream } from "effect"; +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; +import { + type TelegramFile, + TelegramFile as TelegramFileSchema, + type TelegramMessage, + TelegramMessage as TelegramMessageSchema, + type TelegramSendMediaMethod, + type TelegramUser, + TelegramUser as TelegramUserSchema, +} from "./TelegramModels.ts"; + +const TELEGRAM_ADAPTER = AdapterName.make("telegram"); +const DEFAULT_API_BASE_URL = "https://api.telegram.org"; + +export interface TelegramPollingRequest { + readonly offset?: number | undefined; + readonly timeout: number; + readonly limit: number; +} + +export interface TelegramMediaRequest { + readonly method: TelegramSendMediaMethod; + readonly field: "photo" | "audio" | "voice" | "video" | "document"; + readonly chatId: string; + readonly threadId?: number | undefined; + readonly filename: string; + readonly mediaType?: string | undefined; + readonly bytes: Uint8Array; + readonly caption?: string | undefined; +} + +export interface TelegramApiService { + readonly getMe: Effect.Effect; + readonly deleteWebhook: (dropPendingUpdates: boolean) => Effect.Effect; + readonly getUpdates: ( + request: TelegramPollingRequest, + ) => Effect.Effect, GatewayAdapterError>; + readonly getFile: (fileId: string) => Effect.Effect; + readonly download: (filePath: string) => Stream.Stream; + readonly sendText: ( + chatId: string, + threadId: number | undefined, + text: string, + ) => Effect.Effect; + readonly sendMedia: ( + request: TelegramMediaRequest, + ) => Effect.Effect; +} + +export interface TelegramApiOptions { + readonly botToken: Redacted.Redacted; + readonly apiBaseUrl: string; +} + +export class TelegramApi extends Context.Service()( + "@compass/server/gateway/adapters/telegram/TelegramApi", +) { + static readonly layer = Layer.effect( + TelegramApi, + Effect.gen(function* () { + const botToken = yield* Config.redacted("TELEGRAM_BOT_TOKEN"); + const apiBaseUrl = yield* Config.string("TELEGRAM_API_BASE_URL").pipe( + Config.withDefault(DEFAULT_API_BASE_URL), + ); + return yield* makeTelegramApi({ botToken, apiBaseUrl }); + }), + ); + + static readonly layerWith = (options: TelegramApiOptions) => + Layer.effect(TelegramApi, makeTelegramApi(options)); +} + +const apiEnvelope = (result: Schema.ConstraintDecoder) => + Schema.Struct({ + ok: Schema.Boolean, + result: Schema.optionalKey(result), + description: Schema.optionalKey(Schema.String), + error_code: Schema.optionalKey(Schema.Finite), + parameters: Schema.optionalKey( + Schema.Struct({ retry_after: Schema.optionalKey(Schema.Finite) }), + ), + }); + +export const makeTelegramApi = Effect.fn("TelegramApi.make")(function* ( + options: TelegramApiOptions, +) { + const client = yield* HttpClient.HttpClient; + const baseUrl = options.apiBaseUrl.replace(/\/+$/, ""); + const token = Redacted.value(options.botToken); + const methodUrl = (method: string) => `${baseUrl}/bot${token}/${method}`; + + const call = ( + operation: string, + payload: unknown, + result: Schema.ConstraintDecoder, + ): Effect.Effect => + Effect.gen(function* () { + const request = HttpClientRequest.post(methodUrl(operation)).pipe( + HttpClientRequest.bodyJsonUnsafe(payload), + ); + const response = yield* client.execute(request).pipe( + Effect.mapError( + () => + new GatewayTransportError({ + adapter: TELEGRAM_ADAPTER, + operation, + message: `Telegram ${operation} request failed`, + cause: new Error("Telegram HTTP transport failed"), + }), + ), + ); + const json = yield* response.json.pipe( + Effect.mapError( + () => + new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation, + message: `Telegram ${operation} returned an unreadable response`, + cause: new Error("Telegram HTTP response body was unreadable"), + }), + ), + ); + const envelope = yield* Schema.decodeUnknownEffect(apiEnvelope(result))(json).pipe( + Effect.mapError( + (cause) => + new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation, + message: `Telegram ${operation} returned an invalid response`, + cause, + }), + ), + ); + if (!envelope.ok || envelope.result === undefined) { + const retryAfter = envelope.parameters?.retry_after; + const retryDetail = retryAfter === undefined ? "" : `; retry after ${retryAfter}s`; + return yield* new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation, + message: `${envelope.description ?? "Telegram API request failed"}${retryDetail}`, + cause: new Error(`Telegram API error ${envelope.error_code ?? "unknown"}`), + }); + } + return envelope.result; + }); + + const callFormData = Effect.fn("TelegramApi.callFormData")(function* ( + request: TelegramMediaRequest, + ) { + const formData = new FormData(); + formData.set("chat_id", request.chatId); + if (request.threadId !== undefined) { + formData.set("message_thread_id", String(request.threadId)); + } + if (request.caption !== undefined && request.caption.length > 0) { + formData.set("caption", request.caption); + } + formData.set( + request.field, + new Blob([new Uint8Array(request.bytes).buffer], { + type: request.mediaType ?? "application/octet-stream", + }), + request.filename, + ); + const httpRequest = HttpClientRequest.post(methodUrl(request.method)).pipe( + HttpClientRequest.bodyFormData(formData), + ); + const response = yield* client.execute(httpRequest).pipe( + Effect.mapError( + () => + new GatewayTransportError({ + adapter: TELEGRAM_ADAPTER, + operation: request.method, + message: `Telegram ${request.method} request failed`, + cause: new Error("Telegram HTTP transport failed"), + }), + ), + ); + const json = yield* response.json.pipe( + Effect.mapError( + () => + new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation: request.method, + message: `Telegram ${request.method} returned an unreadable response`, + cause: new Error("Telegram HTTP response body was unreadable"), + }), + ), + ); + const envelope = yield* Schema.decodeUnknownEffect(apiEnvelope(TelegramMessageSchema))( + json, + ).pipe( + Effect.mapError( + (cause) => + new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation: request.method, + message: `Telegram ${request.method} returned an invalid response`, + cause, + }), + ), + ); + if (!envelope.ok || envelope.result === undefined) { + return yield* new GatewayProtocolError({ + adapter: TELEGRAM_ADAPTER, + operation: request.method, + message: envelope.description ?? `Telegram ${request.method} failed`, + cause: new Error(`Telegram API error ${envelope.error_code ?? "unknown"}`), + }); + } + return envelope.result; + }); + + const download = (filePath: string) => { + const request = HttpClientRequest.get(`${baseUrl}/file/bot${token}/${filePath}`); + return Stream.unwrap( + client.execute(request).pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.map((response) => + response.stream.pipe( + Stream.mapError( + () => + new GatewayTransportError({ + adapter: TELEGRAM_ADAPTER, + operation: "read file response", + message: "Could not read a Telegram attachment response", + cause: new Error("Telegram attachment response stream failed"), + }), + ), + ), + ), + Effect.mapError( + () => + new GatewayTransportError({ + adapter: TELEGRAM_ADAPTER, + operation: "download file", + message: "Could not download a Telegram attachment", + cause: new Error("Telegram attachment request failed"), + }), + ), + ), + ); + }; + + const deleteWebhook = Effect.fn("TelegramApi.deleteWebhook")(function* ( + dropPendingUpdates: boolean, + ) { + yield* call("deleteWebhook", { drop_pending_updates: dropPendingUpdates }, Schema.Boolean); + }); + + const getUpdates = (request: TelegramPollingRequest) => + call( + "getUpdates", + { + allowed_updates: ["message", "channel_post"], + limit: request.limit, + offset: request.offset, + timeout: request.timeout, + }, + Schema.Array(Schema.Unknown), + ); + + const getFile = (fileId: string) => call("getFile", { file_id: fileId }, TelegramFileSchema); + + const sendText = (chatId: string, threadId: number | undefined, text: string) => + call( + "sendMessage", + { chat_id: chatId, message_thread_id: threadId, text }, + TelegramMessageSchema, + ); + + return TelegramApi.of({ + getMe: call("getMe", {}, TelegramUserSchema), + deleteWebhook, + getUpdates, + getFile, + download, + sendText, + sendMedia: callFormData, + }); +}); diff --git a/apps/server/src/gateway/adapters/telegram/TelegramModels.ts b/apps/server/src/gateway/adapters/telegram/TelegramModels.ts new file mode 100644 index 0000000..db369cd --- /dev/null +++ b/apps/server/src/gateway/adapters/telegram/TelegramModels.ts @@ -0,0 +1,103 @@ +import { Schema } from "effect"; + +const TelegramFileFields = { + file_id: Schema.String, + file_unique_id: Schema.optionalKey(Schema.String), + file_size: Schema.optionalKey(Schema.Finite), +}; + +export const TelegramUser = Schema.Struct({ + id: Schema.Finite, + is_bot: Schema.Boolean, + first_name: Schema.String, + last_name: Schema.optionalKey(Schema.String), + username: Schema.optionalKey(Schema.String), +}); +export type TelegramUser = typeof TelegramUser.Type; + +export const TelegramChat = Schema.Struct({ + id: Schema.Finite, + type: Schema.Literals(["private", "group", "supergroup", "channel"]), + title: Schema.optionalKey(Schema.String), + first_name: Schema.optionalKey(Schema.String), + last_name: Schema.optionalKey(Schema.String), + username: Schema.optionalKey(Schema.String), +}); +export type TelegramChat = typeof TelegramChat.Type; + +export const TelegramFile = Schema.Struct({ + ...TelegramFileFields, + file_path: Schema.optionalKey(Schema.String), +}); +export type TelegramFile = typeof TelegramFile.Type; + +export const TelegramPhotoSize = Schema.Struct({ + ...TelegramFileFields, + width: Schema.Finite, + height: Schema.Finite, +}); +export type TelegramPhotoSize = typeof TelegramPhotoSize.Type; + +const TelegramDocument = Schema.Struct({ + ...TelegramFileFields, + file_name: Schema.optionalKey(Schema.String), + mime_type: Schema.optionalKey(Schema.String), +}); + +const TelegramAudio = Schema.Struct({ + ...TelegramFileFields, + duration: Schema.optionalKey(Schema.Finite), + performer: Schema.optionalKey(Schema.String), + title: Schema.optionalKey(Schema.String), + file_name: Schema.optionalKey(Schema.String), + mime_type: Schema.optionalKey(Schema.String), +}); + +const TelegramVideo = Schema.Struct({ + ...TelegramFileFields, + width: Schema.optionalKey(Schema.Finite), + height: Schema.optionalKey(Schema.Finite), + duration: Schema.optionalKey(Schema.Finite), + file_name: Schema.optionalKey(Schema.String), + mime_type: Schema.optionalKey(Schema.String), +}); + +const TelegramVoice = Schema.Struct({ + ...TelegramFileFields, + duration: Schema.optionalKey(Schema.Finite), + mime_type: Schema.optionalKey(Schema.String), +}); + +export const TelegramMessage = Schema.Struct({ + message_id: Schema.Finite, + message_thread_id: Schema.optionalKey(Schema.Finite), + date: Schema.Finite, + chat: TelegramChat, + from: Schema.optionalKey(TelegramUser), + sender_chat: Schema.optionalKey(TelegramChat), + text: Schema.optionalKey(Schema.String), + caption: Schema.optionalKey(Schema.String), + photo: Schema.optionalKey(Schema.Array(TelegramPhotoSize)), + audio: Schema.optionalKey(TelegramAudio), + document: Schema.optionalKey(TelegramDocument), + video: Schema.optionalKey(TelegramVideo), + video_note: Schema.optionalKey(TelegramVideo), + voice: Schema.optionalKey(TelegramVoice), +}); +export type TelegramMessage = typeof TelegramMessage.Type; + +export const TelegramUpdateIdentity = Schema.Struct({ update_id: Schema.Finite }); + +export const TelegramUpdate = Schema.Struct({ + update_id: Schema.Finite, + message: Schema.optionalKey(TelegramMessage), + channel_post: Schema.optionalKey(TelegramMessage), +}); +export type TelegramUpdate = typeof TelegramUpdate.Type; + +export type TelegramSendMediaMethod = + | "sendPhoto" + | "sendAudio" + | "sendVoice" + | "sendVideo" + | "sendDocument"; diff --git a/apps/server/src/gateway/adapters/telegram/index.ts b/apps/server/src/gateway/adapters/telegram/index.ts new file mode 100644 index 0000000..1d0a271 --- /dev/null +++ b/apps/server/src/gateway/adapters/telegram/index.ts @@ -0,0 +1,3 @@ +export * from "./TelegramAdapter.ts"; +export * from "./TelegramApi.ts"; +export * from "./TelegramModels.ts"; diff --git a/apps/server/src/gateway/index.ts b/apps/server/src/gateway/index.ts new file mode 100644 index 0000000..69a6bfc --- /dev/null +++ b/apps/server/src/gateway/index.ts @@ -0,0 +1,72 @@ +import { NodeServices } from "@effect/platform-node"; +import { Effect, Layer } from "effect"; +import { FetchHttpClient } from "effect/unstable/http"; +import { Gateway, makeGateway } from "./Gateway.ts"; +import { GatewayConfig, layerDefault as gatewayConfigLayerDefault } from "./GatewayConfig.ts"; +import { layerDefault as gatewayFilesLayerDefault } from "./GatewayFiles.ts"; +import { TelegramAdapter } from "./adapters/telegram/TelegramAdapter.ts"; +import { TelegramApi } from "./adapters/telegram/TelegramApi.ts"; + +export { + AdapterName, + ChannelId, + type GatewayAdapter, + type GatewayAdapterError, + GatewayAdapterNotFoundError, + GatewayAuthorizationError, + GatewayConfigurationError, + type GatewayError, + GatewayFile, + GatewayFileError, + GatewayFileKind, + GatewayInvalidMessageError, + GatewayMessageId, + GatewayProtocolError, + GatewayTimestampMillis, + GatewayTransportError, + type IncomingAdapterEvent, + IncomingMessage, + MessageSender, + OutgoingMessage, + type ParsedChannelId, + SentMessage, + ThreadId, + parseChannelId, +} from "@compass/contracts"; +export * from "./Gateway.ts"; +export { + GatewayConfig, + type GatewayConfigFileOptions, + type GatewaySettings, + type TelegramGatewaySettings, + layerDefault as gatewayConfigLayerDefault, + layerFile as gatewayConfigLayerFile, + loadGatewayConfig, +} from "./GatewayConfig.ts"; +export { + GatewayFiles, + type GatewayFilesService, + type SaveGatewayFileOptions, + type SaveGatewayFileStreamOptions, + layerAt as gatewayFilesLayerAt, + layerDefault as gatewayFilesLayerDefault, +} from "./GatewayFiles.ts"; +export * as Telegram from "./adapters/telegram/index.ts"; + +export const telegramGatewayLayer = Layer.effect( + Gateway, + Effect.gen(function* () { + const config = yield* GatewayConfig; + const telegram = yield* TelegramAdapter; + return yield* makeGateway([telegram], config); + }), +); + +export const telegramGatewayLayerLive = telegramGatewayLayer.pipe( + Layer.provide(TelegramAdapter.layer), + Layer.provide(TelegramApi.layer), + Layer.provide(gatewayConfigLayerDefault), + Layer.provide(gatewayFilesLayerDefault), + Layer.provide(FetchHttpClient.layer), + Layer.provide(NodeServices.layer), +); diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 3aeacc0..6236fcb 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -1,5 +1,6 @@ import { Effect } from "effect"; export * from "./harness/index.ts"; +export * from "./gateway/index.ts"; export const main = Effect.void; diff --git a/packages/contracts/src/gateway.test.ts b/packages/contracts/src/gateway.test.ts new file mode 100644 index 0000000..2b8ce2c --- /dev/null +++ b/packages/contracts/src/gateway.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Option, Schema } from "effect"; +import { + ChannelId, + GatewayFile, + GatewayTimestampMillis, + IncomingMessage, + GatewayMessageId, + MessageSender, + parseChannelId, +} from "./gateway.ts"; + +describe("gateway contracts", () => { + it.effect("decodes adapter-qualified channels and gateway timestamps", () => + Effect.gen(function* () { + const channel = yield* Schema.decodeUnknownEffect(ChannelId)("telegram:123"); + const receivedAt = yield* Schema.decodeUnknownEffect(GatewayTimestampMillis)(1_000); + + expect(parseChannelId(channel)).toEqual({ adapter: "telegram", nativeChannel: "123" }); + expect(receivedAt).toBe(1_000); + }), + ); + + it.effect("rejects malformed gateway identifiers", () => + Effect.gen(function* () { + expect( + Option.isNone(yield* Effect.option(Schema.decodeUnknownEffect(ChannelId)("123"))), + ).toBe(true); + expect( + Option.isNone(yield* Effect.option(Schema.decodeUnknownEffect(GatewayTimestampMillis)(-1))), + ).toBe(true); + }), + ); + + it("keeps incoming messages and files schema-backed", () => { + const message = new IncomingMessage({ + id: GatewayMessageId.make("1"), + channel: ChannelId.make("telegram:123"), + sender: new MessageSender({ id: "7", displayName: "Ada", isBot: false }), + files: [new GatewayFile({ path: "C:/files/photo.jpg", name: "photo.jpg", kind: "image" })], + receivedAt: GatewayTimestampMillis.make(1_000), + }); + + expect(message.files[0]?.kind).toBe("image"); + }); +}); diff --git a/packages/contracts/src/gateway.ts b/packages/contracts/src/gateway.ts new file mode 100644 index 0000000..e8cc505 --- /dev/null +++ b/packages/contracts/src/gateway.ts @@ -0,0 +1,162 @@ +import { type Effect, Schema, type Stream } from "effect"; + +const NonEmptyString = Schema.String.check(Schema.isMinLength(1)); +const NonNegativeInteger = Schema.Finite.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(0)); +const QUALIFIED_CHANNEL_PATTERN = /^[a-z][a-z0-9-]*:.+$/; + +export const AdapterName = NonEmptyString.pipe(Schema.brand("GatewayAdapterName")); +export type AdapterName = typeof AdapterName.Type; + +export const ChannelId = Schema.String.check(Schema.isPattern(QUALIFIED_CHANNEL_PATTERN)).pipe( + Schema.brand("GatewayChannelId"), +); +export type ChannelId = typeof ChannelId.Type; + +export const GatewayMessageId = NonEmptyString.pipe(Schema.brand("GatewayMessageId")); +export type GatewayMessageId = typeof GatewayMessageId.Type; + +export const ThreadId = NonEmptyString.pipe(Schema.brand("GatewayThreadId")); +export type ThreadId = typeof ThreadId.Type; + +export const GatewayTimestampMillis = NonNegativeInteger.pipe( + Schema.brand("GatewayTimestampMillis"), +); +export type GatewayTimestampMillis = typeof GatewayTimestampMillis.Type; + +export const GatewayFileKind = Schema.Literals(["image", "audio", "video", "file"]); +export type GatewayFileKind = typeof GatewayFileKind.Type; + +export class GatewayFile extends Schema.Class( + "@compass/contracts/gateway/GatewayFile", +)({ + path: NonEmptyString, + name: NonEmptyString, + kind: GatewayFileKind, + mediaType: Schema.optionalKey(NonEmptyString), + size: Schema.optionalKey(NonNegativeInteger), +}) {} + +export class MessageSender extends Schema.Class( + "@compass/contracts/gateway/MessageSender", +)({ + id: NonEmptyString, + displayName: NonEmptyString, + username: Schema.optionalKey(NonEmptyString), + isBot: Schema.Boolean, +}) {} + +export class IncomingMessage extends Schema.Class( + "@compass/contracts/gateway/IncomingMessage", +)({ + id: GatewayMessageId, + channel: ChannelId, + thread: Schema.optionalKey(ThreadId), + sender: MessageSender, + text: Schema.optionalKey(Schema.String), + files: Schema.Array(GatewayFile), + receivedAt: GatewayTimestampMillis, +}) {} + +export class OutgoingMessage extends Schema.Class( + "@compass/contracts/gateway/OutgoingMessage", +)({ + channel: ChannelId, + thread: Schema.optionalKey(ThreadId), + text: Schema.optionalKey(Schema.String), + files: Schema.Array(GatewayFile), +}) {} + +export class SentMessage extends Schema.Class( + "@compass/contracts/gateway/SentMessage", +)({ + channel: ChannelId, + thread: Schema.optionalKey(ThreadId), + messageIds: Schema.Array(GatewayMessageId), +}) {} + +export class GatewayConfigurationError extends Schema.TaggedErrorClass()( + "GatewayConfigurationError", + { + path: Schema.String, + message: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export class GatewayAuthorizationError extends Schema.TaggedErrorClass()( + "GatewayAuthorizationError", + { channel: ChannelId }, +) {} + +export class GatewayAdapterNotFoundError extends Schema.TaggedErrorClass()( + "GatewayAdapterNotFoundError", + { adapter: AdapterName, channel: ChannelId }, +) {} + +export class GatewayInvalidMessageError extends Schema.TaggedErrorClass()( + "GatewayInvalidMessageError", + { channel: ChannelId, message: Schema.String }, +) {} + +export class GatewayFileError extends Schema.TaggedErrorClass()( + "GatewayFileError", + { + operation: NonEmptyString, + path: Schema.String, + message: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export class GatewayTransportError extends Schema.TaggedErrorClass()( + "GatewayTransportError", + { + adapter: AdapterName, + operation: NonEmptyString, + message: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export class GatewayProtocolError extends Schema.TaggedErrorClass()( + "GatewayProtocolError", + { + adapter: AdapterName, + operation: NonEmptyString, + message: Schema.String, + cause: Schema.Defect(), + }, +) {} + +export type GatewayAdapterError = GatewayFileError | GatewayTransportError | GatewayProtocolError; + +export type GatewayError = + | GatewayAdapterError + | GatewayConfigurationError + | GatewayAuthorizationError + | GatewayAdapterNotFoundError + | GatewayInvalidMessageError; + +export interface ParsedChannelId { + readonly adapter: AdapterName; + readonly nativeChannel: string; +} + +export const parseChannelId = (channel: ChannelId): ParsedChannelId => { + const separator = channel.indexOf(":"); + return { + adapter: AdapterName.make(channel.slice(0, separator)), + nativeChannel: channel.slice(separator + 1), + }; +}; + +export interface IncomingAdapterEvent { + readonly channel: ChannelId; + readonly materialize: Effect.Effect; +} + +export interface GatewayAdapter { + readonly name: AdapterName; + readonly incoming: Stream.Stream; + readonly send: (message: OutgoingMessage) => Effect.Effect; +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index d8221fc..8b4866b 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -1,2 +1,3 @@ export * as Schema from "effect/Schema"; +export * from "./gateway.ts"; export * from "./harness.ts";