Run a Vercel eve agent over iMessage

/**
 * Comms by Osis — iMessage/SMS channel for Vercel eve.
 *
 * Bridges Comms' iMessage infrastructure (comms.osis.co) to the eve agent
 * runtime. Comms fires a `comms.message.received` webhook at POST /webhook,
 * the channel starts or resumes a durable session keyed by the Comms
 * conversation id, and the agent's reply is delivered back over iMessage via
 * POST {COMMS_API_URL}/messages.
 *
 * Env:
 *   COMMS_API_KEY            bearer token with the comms_send scope (outbound)
 *   COMMS_API_URL            base URL, default https://osis.co/api/v1/comms
 *   COMMS_WEBHOOK_SECRET     whsec_… endpoint secret from webhook registration
 *   IMESSAGE_ALLOW_FROM      comma-separated E.164 allowlist, or "*"
 *   IMESSAGE_DEFAULT_CHANNEL "imessage" | "sms" (default "imessage")
 */

import { createHash, createHmac, timingSafeEqual } from "node:crypto";
import {
  defineChannel,
  POST,
  type RouteHelpers,
  type Session,
  type UserContent,
  type UserContentPart,
} from "eve/channels";

// ── Types ───────────────────────────────────────────────────────────────

export type ImessageWire = "imessage" | "sms";

export interface InboundAttachment {
  url: string;
  mediaType: string;
}

/** One inbound message, normalized from either Comms payload shape. */
export interface InboundImessage {
  /** Sender phone, E.164. Becomes auth.principalId. */
  from: string;
  /** Receiving line (the Comms-managed number), when the payload carries it. */
  to: string | null;
  body: string;
  /** Conversation identifier: Comms conversation_id, chat GUID, or from:to. */
  chatId: string;
  /** The Comms conversation_id when the payload carries one — valid as an outbound reply target. */
  conversationId: string | null;
  messageId: string | null;
  messageType: ImessageWire;
  senderName: string | null;
  attachments: InboundAttachment[];
  timestamp: string | null;
}

export type AllowFrom =
  | string
  | string[]
  | ((from: string, message: InboundImessage) => boolean | Promise<boolean>);

export interface ImessageChannelOptions {
  /**
   * Who may talk to the agent: a single E.164 number, a list, "*" (open to
   * anyone — dangerous), or an async resolver. Defaults to the
   * IMESSAGE_ALLOW_FROM env var; with no configuration the channel denies
   * everyone rather than defaulting open.
   */
  allowFrom?: AllowFrom;
  /**
   * Inspect each inbound message before dispatch. Return null to drop it, or
   * an object (optionally overriding auth attributes) to proceed.
   */
  onText?: (
    message: InboundImessage,
  ) =>
    | { attributes?: Record<string, unknown> }
    | null
    | Promise<{ attributes?: Record<string, unknown> } | null>;
  apiKey?: string;
  apiUrl?: string;
  webhookSecret?: string;
  defaultChannel?: ImessageWire;
}

/** Per-session context built by context(); event handlers receive it. */
export interface ImessageSessionContext {
  sessionId: string | null;
  replyPhone: string | null;
  conversationId: string | null;
  chatId: string | null;
  wire: ImessageWire;
  senderName: string | null;
}

interface ResolvedConfig {
  apiKey: string;
  apiUrl: string;
  webhookSecret: string;
  defaultChannel: ImessageWire;
  allowFrom: AllowFrom | null;
  onText: ImessageChannelOptions["onText"] | null;
}

// ── Config ──────────────────────────────────────────────────────────────

const DEFAULT_API_URL = "https://osis.co/api/v1/comms";
const MAX_ATTACHMENTS = 8;
const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024;
const NEW_SESSION_COMMANDS = new Set(["/new", "/reset"]);

function resolveConfig(options: ImessageChannelOptions): ResolvedConfig {
  const env = process.env;
  const defaultChannel =
    options.defaultChannel ??
    (env.IMESSAGE_DEFAULT_CHANNEL === "sms" ? "sms" : "imessage");
  return {
    apiKey: options.apiKey ?? env.COMMS_API_KEY ?? "",
    apiUrl: (options.apiUrl ?? env.COMMS_API_URL ?? DEFAULT_API_URL).replace(/\/+$/, ""),
    webhookSecret: options.webhookSecret ?? env.COMMS_WEBHOOK_SECRET ?? "",
    defaultChannel,
    allowFrom: options.allowFrom ?? env.IMESSAGE_ALLOW_FROM ?? null,
    onText: options.onText ?? null,
  };
}

/**
 * The raw continuation token for a Comms conversation. The framework
 * namespaces it with the channel id (`imessage:<token>`); use this helper
 * when targeting the channel from elsewhere (e.g. helpers.receive()).
 */
export function imessageContinuationToken(chatId: string): string {
  return chatId.trim();
}

// ── Signature verification ──────────────────────────────────────────────

function verifySignature(rawBody: string, header: string | null, secret: string): boolean {
  if (!header) return false;
  const provided = header.startsWith("sha256=") ? header.slice(7) : header;
  if (!/^[0-9a-f]{64}$/i.test(provided)) return false;
  const expected = createHmac("sha256", secret).update(rawBody).digest();
  const given = Buffer.from(provided, "hex");
  return given.length === expected.length && timingSafeEqual(given, expected);
}

// ── Duplicate suppression ───────────────────────────────────────────────
// Comms retries undelivered webhooks up to 6 times with backoff, so the same
// message can arrive more than once. Best-effort per-instance LRU; a retry
// that lands on a cold instance re-runs, which is safe (same continuation
// token, same session).

const seenMessages = new Map<string, true>();
const SEEN_CAPACITY = 512;

function markSeen(key: string): boolean {
  if (seenMessages.has(key)) return true;
  seenMessages.set(key, true);
  if (seenMessages.size > SEEN_CAPACITY) {
    const oldest = seenMessages.keys().next().value;
    if (oldest !== undefined) seenMessages.delete(oldest);
  }
  return false;
}

function forgetSeen(key: string): void {
  seenMessages.delete(key);
}

// ── Payload parsing ─────────────────────────────────────────────────────

type ParsedWebhook =
  | { kind: "message"; message: InboundImessage }
  | { kind: "ignored"; reason: string }
  | { kind: "ping" };

function asRecord(value: unknown): Record<string, unknown> {
  return typeof value === "object" && value !== null ? (value as Record<string, unknown>) : {};
}

function str(value: unknown): string | null {
  return typeof value === "string" && value.length > 0 ? value : null;
}

function parseAttachments(value: unknown): InboundAttachment[] {
  if (!Array.isArray(value)) return [];
  const attachments: InboundAttachment[] = [];
  for (const entry of value) {
    const record = asRecord(entry);
    const url = str(record.url);
    if (!url) continue;
    attachments.push({ url, mediaType: str(record.mediaType) ?? "application/octet-stream" });
  }
  return attachments;
}

function wireOf(value: unknown, fallback: ImessageWire): ImessageWire {
  return value === "sms" ? "sms" : value === "imessage" ? "imessage" : fallback;
}

/**
 * Accepts both payload shapes:
 *  - the Osis webhook envelope: { event, object, recordId, data: { message,
 *    contact, conversation_id }, sentAt } — what production Comms sends today
 *  - the flat InboundMessage: { from, to, body, chatId, attachments, ... }
 */
function parseWebhook(payload: unknown, defaultChannel: ImessageWire): ParsedWebhook {
  const root = asRecord(payload);

  const event = str(root.event);
  if (event?.startsWith("comms.")) {
    if (event === "comms.ping") return { kind: "ping" };
    if (event !== "comms.message.received") {
      return { kind: "ignored", reason: `event ${event}` };
    }
    const data = asRecord(root.data);
    const message = asRecord(data.message);
    const contact = asRecord(data.contact);
    if (message.direction !== "inbound") {
      return { kind: "ignored", reason: "not an inbound message" };
    }
    const channel = str(message.channel) ?? "sms";
    if (channel !== "imessage" && channel !== "sms") {
      return { kind: "ignored", reason: `channel ${channel}` };
    }
    const from = str(contact.phone);
    if (!from) return { kind: "ignored", reason: "contact has no phone" };
    const chatId = str(data.conversation_id) ?? str(message.conversation_id);
    if (!chatId) return { kind: "ignored", reason: "no conversation id" };
    return {
      kind: "message",
      message: {
        from,
        to: str(message.to) ?? str(data.to),
        body: str(message.body) ?? "",
        chatId,
        conversationId: chatId,
        messageId: str(message.id) ?? str(root.recordId),
        messageType: channel,
        senderName: str(contact.name),
        attachments: parseAttachments(message.attachments ?? data.attachments),
        timestamp: str(message.created_at) ?? str(root.sentAt),
      },
    };
  }

  const from = str(root.from);
  if (from && (typeof root.body === "string" || Array.isArray(root.attachments))) {
    const to = str(root.to);
    return {
      kind: "message",
      message: {
        from,
        to,
        body: str(root.body) ?? "",
        chatId: str(root.chatId) ?? (to ? `${from}:${to}` : from),
        // chatId may be an iMessage GUID rather than a Comms conversation id,
        // so only an explicit conversationId qualifies as a reply target.
        conversationId: str(root.conversationId) ?? str(root.conversation_id),
        messageId: str(root.messageId) ?? str(root.id),
        messageType: wireOf(root.messageType, defaultChannel),
        senderName: str(root.senderName),
        attachments: parseAttachments(root.attachments),
        timestamp: str(root.timestamp),
      },
    };
  }

  return { kind: "ignored", reason: "unrecognized payload shape" };
}

// ── Allowlist ───────────────────────────────────────────────────────────

function phoneKey(value: string): string {
  return value.replace(/[^0-9]/g, "");
}

async function isAllowed(allow: AllowFrom | null, message: InboundImessage): Promise<boolean> {
  if (allow === null) {
    console.warn(
      "[imessage-channel] IMESSAGE_ALLOW_FROM is not set; dropping inbound message. " +
        'Set an allowlist (or "*" to allow anyone).',
    );
    return false;
  }
  if (typeof allow === "function") return Boolean(await allow(message.from, message));
  const entries = (Array.isArray(allow) ? allow : allow.split(","))
    .map((entry) => entry.trim())
    .filter(Boolean);
  if (entries.includes("*")) return true;
  const from = phoneKey(message.from);
  return entries.some((entry) => phoneKey(entry) === from);
}

// ── Comms outbound client ───────────────────────────────────────────────

interface CommsSendInput {
  to?: string;
  conversationId?: string;
  body: string;
  channel: ImessageWire;
  idempotencyKey?: string;
}

async function commsSend(config: ResolvedConfig, input: CommsSendInput): Promise<void> {
  if (!config.apiKey) {
    console.error("[imessage-channel] COMMS_API_KEY is not set; cannot deliver reply");
    return;
  }
  const payload: Record<string, unknown> = {
    body: input.body,
    channel: input.channel,
  };
  // conversation_id keeps the reply in the same Comms conversation (correct
  // for group chats); fall back to a direct send by phone.
  if (input.conversationId) payload.conversation_id = input.conversationId;
  else payload.to = input.to;
  if (input.idempotencyKey) payload.idempotency_key = input.idempotencyKey;

  const maxAttempts = 3;
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    let status = 0;
    let detail = "";
    try {
      const response = await fetch(`${config.apiUrl}/messages`, {
        method: "POST",
        headers: {
          authorization: `Bearer ${config.apiKey}`,
          "content-type": "application/json",
        },
        body: JSON.stringify(payload),
        signal: AbortSignal.timeout(15_000),
      });
      if (response.ok) return;
      status = response.status;
      detail = (await response.text().catch(() => "")).slice(0, 300);
      // A stale or foreign conversation id 404s — fall back to a direct send.
      if (status === 404 && input.conversationId && input.to) {
        return commsSend(config, { ...input, conversationId: undefined });
      }
      // 4xx (other than rate limiting) will not succeed on retry.
      if (status < 500 && status !== 429) {
        console.error(`[imessage-channel] Comms send rejected (http ${status}): ${detail}`);
        return;
      }
      if (attempt < maxAttempts) {
        const retryAfter = Number(response.headers.get("retry-after"));
        const delayMs =
          Number.isFinite(retryAfter) && retryAfter > 0
            ? Math.min(retryAfter * 1000, 30_000)
            : 500 * 2 ** (attempt - 1);
        await new Promise((resolve) => setTimeout(resolve, delayMs));
        continue;
      }
    } catch (error) {
      detail = error instanceof Error ? error.message : String(error);
      if (attempt < maxAttempts) {
        await new Promise((resolve) => setTimeout(resolve, 500 * 2 ** (attempt - 1)));
        continue;
      }
    }
    console.error(
      `[imessage-channel] Comms send failed after ${maxAttempts} attempts` +
        (status ? ` (http ${status})` : "") +
        (detail ? `: ${detail}` : ""),
    );
  }
}

function deliveryTarget(context: Partial<ImessageSessionContext> | undefined): {
  to?: string;
  conversationId?: string;
  channel: ImessageWire;
} | null {
  if (!context) return null;
  const conversationId = context.conversationId ?? undefined;
  const to = context.replyPhone ?? undefined;
  if (!conversationId && !to) return null;
  return { conversationId, to, channel: context.wire === "sms" ? "sms" : "imessage" };
}

function replyIdempotencyKey(
  context: Partial<ImessageSessionContext> | undefined,
  eventData: Record<string, unknown>,
  text: string,
): string | undefined {
  const sessionId = context?.sessionId;
  if (!sessionId) return undefined;
  const turn = str(eventData.turnId) ?? str(eventData.messageId) ?? "";
  const digest = createHash("sha256").update(text).digest("hex").slice(0, 16);
  return `eve:${sessionId}:${turn || digest}`;
}

// ── Channel ─────────────────────────────────────────────────────────────

export function createImessageChannel(options: ImessageChannelOptions = {}) {
  const config = resolveConfig(options);

  async function handleInbound(message: InboundImessage, helpers: RouteHelpers): Promise<void> {
    const continuationToken = imessageContinuationToken(message.chatId);

    if (NEW_SESSION_COMMANDS.has(message.body.trim().toLowerCase())) {
      const result = await helpers.reset({
        continuationToken,
        reason: `user sent ${message.body.trim()}`,
      });
      helpers.waitUntil(
        commsSend(config, {
          conversationId: message.conversationId ?? undefined,
          to: message.from,
          body:
            result.status === "reset"
              ? "Started a new conversation — previous context is cleared."
              : "You're already in a fresh conversation.",
          channel: message.messageType,
          idempotencyKey: message.messageId ? `eve:reset:${message.messageId}` : undefined,
        }),
      );
      return;
    }

    let attributes: Record<string, unknown> = {
      chatId: message.chatId,
      conversationId: message.conversationId,
      to: message.to,
      replyTo: message.from,
      channel: message.messageType,
      senderName: message.senderName,
    };
    if (config.onText) {
      const decision = await config.onText(message);
      if (!decision) return;
      if (decision.attributes) attributes = { ...attributes, ...decision.attributes };
    }

    const parts: UserContentPart[] = [];
    if (message.body) parts.push({ type: "text", text: message.body });
    for (const attachment of message.attachments.slice(0, MAX_ATTACHMENTS)) {
      parts.push({
        type: "file",
        data: new URL(attachment.url),
        mediaType: attachment.mediaType,
      });
    }
    if (message.attachments.length > MAX_ATTACHMENTS) {
      console.warn(
        `[imessage-channel] dropped ${message.attachments.length - MAX_ATTACHMENTS} ` +
          `attachment(s) over the ${MAX_ATTACHMENTS}-per-message cap`,
      );
    }
    const content: UserContent =
      parts.length === 1 && parts[0]?.type === "text" ? message.body : parts;

    await helpers.send(content, {
      auth: {
        principalId: message.from,
        principalType: "user",
        authenticator: "comms-imessage",
        attributes,
      },
      continuationToken,
      title: message.senderName ?? message.from,
    });
  }

  return defineChannel<Record<string, never>, ImessageSessionContext>({
    context(_state, session: Session) {
      const attributes = asRecord(session?.auth?.attributes);
      return {
        sessionId: session?.id ?? null,
        replyPhone: str(attributes.replyTo) ?? session?.auth?.principalId ?? null,
        conversationId: str(attributes.conversationId),
        chatId: str(attributes.chatId),
        wire: wireOf(attributes.channel, config.defaultChannel),
        senderName: str(attributes.senderName),
      };
    },

    // iMessage attachments live behind Comms-authenticated URLs; attach the
    // API key only for same-origin fetches so it never leaks to other hosts.
    async fetchFile(url) {
      const target = new URL(url);
      const headers: Record<string, string> = {};
      if (config.apiKey && target.origin === new URL(config.apiUrl).origin) {
        headers.authorization = `Bearer ${config.apiKey}`;
      }
      const response = await fetch(target, { headers, signal: AbortSignal.timeout(30_000) });
      if (!response.ok) {
        throw new Error(`attachment fetch failed (http ${response.status}): ${target.pathname}`);
      }
      const length = Number(response.headers.get("content-length") ?? 0);
      if (length > MAX_ATTACHMENT_BYTES) {
        throw new Error(
          `attachment too large (${length} bytes > ${MAX_ATTACHMENT_BYTES} byte cap)`,
        );
      }
      return response;
    },

    routes: [
      POST("/webhook", async (req, helpers) => {
        if (!config.webhookSecret) {
          console.error("[imessage-channel] COMMS_WEBHOOK_SECRET is not set; rejecting webhook");
          return Response.json({ error: "webhook secret not configured" }, { status: 500 });
        }
        const rawBody = await req.text();
        const signature = req.headers.get("x-osis-signature");
        if (!verifySignature(rawBody, signature, config.webhookSecret)) {
          return Response.json({ error: "invalid signature" }, { status: 401 });
        }

        let payload: unknown;
        try {
          payload = JSON.parse(rawBody);
        } catch {
          return Response.json({ error: "invalid json" }, { status: 400 });
        }

        const parsed = parseWebhook(payload, config.defaultChannel);
        if (parsed.kind === "ping") return Response.json({ ok: true, pong: true });
        if (parsed.kind === "ignored") {
          return Response.json({ ok: true, ignored: parsed.reason });
        }

        const message = parsed.message;
        const dedupeKey =
          message.messageId ??
          createHash("sha256")
            .update(`${message.from}:${message.body}:${message.timestamp ?? ""}`)
            .digest("hex");
        if (markSeen(dedupeKey)) {
          return Response.json({ ok: true, duplicate: true });
        }

        if (!(await isAllowed(config.allowFrom, message))) {
          return Response.json({ ok: true, dropped: "sender not allowed" });
        }
        if (!message.body.trim() && message.attachments.length === 0) {
          return Response.json({ ok: true, dropped: "empty message" });
        }

        try {
          await handleInbound(message, helpers);
        } catch (error) {
          // Undo the dedupe claim so Comms' webhook retry gets a clean run.
          forgetSeen(dedupeKey);
          console.error(
            "[imessage-channel] inbound dispatch failed:",
            error instanceof Error ? error.message : error,
          );
          return Response.json({ error: "dispatch failed" }, { status: 500 });
        }
        return Response.json({ ok: true }, { status: 202 });
      }),
    ],

    events: {
      "message.completed"(eventData, channel) {
        const text = (str(asRecord(eventData).text) ?? "").trim();
        const target = deliveryTarget(channel);
        if (!text || !target) {
          if (!target) console.error("[imessage-channel] no delivery target for completed message");
          return;
        }
        void commsSend(config, {
          ...target,
          body: text,
          idempotencyKey: replyIdempotencyKey(channel, asRecord(eventData), text),
        });
      },

      // iMessage has no buttons: render human-in-the-loop approval as plain
      // text. The user's next reply flows into the same session via the same
      // continuation token, which resolves the pending input request.
      "input.requested"(eventData, channel) {
        const prompt = (str(asRecord(eventData).prompt) ?? "").trim();
        const target = deliveryTarget(channel);
        if (!prompt || !target) return;
        void commsSend(config, {
          ...target,
          body: prompt,
          idempotencyKey: replyIdempotencyKey(channel, asRecord(eventData), `input:${prompt}`),
        });
      },

      "session.failed"(eventData, channel) {
        const reason = str(asRecord(eventData).error) ?? str(asRecord(eventData).reason) ?? "unknown";
        console.error(`[imessage-channel] session failed: ${reason}`);
        const target = deliveryTarget(channel);
        if (!target) return;
        void commsSend(config, {
          ...target,
          body: "Sorry — something went wrong on my end. Text /new to start fresh.",
        });
      },
    },
  });
}

export default createImessageChannel();