openclaw/extensions/nextcloud-talk/src/inbound.ts

import { resolveChannelInboundRouteEnvelope } from "openclaw/plugin-sdk/channel-inbound";
// Nextcloud Talk plugin module implements inbound behavior.
import {
  channelIngressRoutes,
  resolveStableChannelMessageIngress,
  type ChannelIngressContextBinding,
} from "openclaw/plugin-sdk/channel-ingress-runtime";
import {
  bindIngressLifecycleToReplyOptions,
  resolveChannelStreamingBlockEnabled,
} from "openclaw/plugin-sdk/channel-outbound";
import {
  normalizeOptionalString,
  normalizeStringEntries,
} from "openclaw/plugin-sdk/string-coerce-runtime";
import { sanitizeAssistantVisibleText } from "openclaw/plugin-sdk/text-chunking";
import {
  GROUP_POLICY_BLOCKED_LABEL,
  resolveAllowlistProviderRuntimeGroupPolicy,
  createChannelPairingController,
  deliverFormattedTextWithAttachments,
  logInboundDrop,
  resolveDefaultGroupPolicy,
  warnMissingProviderGroupPolicyFallbackOnce,
  type GroupPolicy,
  type OpenClawConfig,
  type OutboundReplyPayload,
  type RuntimeEnv,
} from "../runtime-api.js";
import type { ResolvedNextcloudTalkAccount } from "./accounts.js";
import {
  normalizeNextcloudTalkAllowEntry,
  normalizeNextcloudTalkAllowlist,
  resolveNextcloudTalkGroupRequireMention,
  resolveNextcloudTalkAllowlistMatch,
  resolveNextcloudTalkRoomMatch,
} from "./policy.js";
import { resolveNextcloudTalkRoomKind } from "./room-info.js";
import { getNextcloudTalkRuntime } from "./runtime.js";
import { sendMessageNextcloudTalk } from "./send.js";
import type { CoreConfig, NextcloudTalkInboundMessage, NextcloudTalkRoomConfig } from "./types.js";

const CHANNEL_ID = "nextcloud-talk" as const;

type NextcloudTalkRoomMatch = ReturnType<typeof resolveNextcloudTalkRoomMatch>;

function hasAllowEntries(entries: string[]): boolean {
  return normalizeNextcloudTalkAllowlist(entries).length > 0;
}

function roomRoutes(params: {
  isGroup: boolean;
  groupPolicy: GroupPolicy;
  roomMatch: NextcloudTalkRoomMatch;
  roomConfig?: NextcloudTalkRoomConfig;
  senderId: string;
  outerGroupAllowFrom: string[];
  roomAllowFrom: string[];
}) {
  if (!params.isGroup) {
    return [];
  }
  const roomSenderConfigured =
    params.groupPolicy === "allowlist" && hasAllowEntries(params.roomAllowFrom);
  return channelIngressRoutes(
    params.roomMatch.allowlistConfigured && {
      id: "nextcloud-talk:room",
      allowed: params.roomMatch.allowed,
      precedence: 0,
      matchId: "nextcloud-talk-room",
      blockReason: "room_not_allowlisted",
    },
    params.roomConfig?.enabled === false && {
      id: "nextcloud-talk:room-enabled",
      enabled: false,
      precedence: 10,
      blockReason: "room_disabled",
    },
    roomSenderConfigured && {
      id: "nextcloud-talk:room-sender",
      kind: "nestedAllowlist",
      precedence: 20,
      blockReason: "room_sender_not_allowlisted",
      ...(!hasAllowEntries(params.outerGroupAllowFrom)
        ? {
            senderPolicy: "replace" as const,
            senderAllowFrom: params.roomAllowFrom,
          }
        : {
            allowed: resolveNextcloudTalkAllowlistMatch({
              allowFrom: params.roomAllowFrom,
              senderId: params.senderId,
            }).allowed,
            matchId: "nextcloud-talk-room-sender",
          }),
    },
  );
}

async function deliverNextcloudTalkReply(params: {
  cfg: CoreConfig;
  payload: OutboundReplyPayload;
  roomToken: string;
  accountId: string;
  statusSink?: (patch: { lastOutboundAt?: number }) => void;
}): Promise<{ visibleReplySent: boolean }> {
  const { cfg, payload, roomToken, accountId, statusSink } = params;
  const visibleReplySent = await deliverFormattedTextWithAttachments({
    payload,
    send: async ({ text, replyToId }) => {
      await sendMessageNextcloudTalk(roomToken, text, {
        cfg,
        accountId,
        replyTo: replyToId,
      });
      statusSink?.({ lastOutboundAt: Date.now() });
    },
  });
  return { visibleReplySent };
}

export async function handleNextcloudTalkInbound(params: {
  message: NextcloudTalkInboundMessage;
  account: ResolvedNextcloudTalkAccount;
  config: CoreConfig;
  runtime: RuntimeEnv;
  statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
  turnAdoptionLifecycle?: Parameters<typeof bindIngressLifecycleToReplyOptions>[0];
}): Promise<void> {
  const { message, account, config, runtime, statusSink } = params;
  const core = getNextcloudTalkRuntime();
  const pairing = createChannelPairingController({
    core,
    channel: CHANNEL_ID,
    accountId: account.accountId,
  });

  const rawBody = message.text?.trim() ?? "";
  if (!rawBody) {
    return;
  }

  const roomKind = await resolveNextcloudTalkRoomKind({
    account,
    roomToken: message.roomToken,
    runtime,
  });
  const isGroup = roomKind === "direct" ? false : roomKind === "group" ? true : message.isGroupChat;
  const senderId = message.senderId;
  const senderName = message.senderName;
  const roomToken = message.roomToken;
  const roomName = message.roomName;

  statusSink?.({ lastInboundAt: message.timestamp });

  const roomMatch = resolveNextcloudTalkRoomMatch({
    rooms: account.config.rooms,
    roomToken,
  });
  const roomConfig = roomMatch.roomConfig;
  const allowTextCommands = core.channel.commands.shouldHandleTextCommands({
    cfg: config as OpenClawConfig,
    surface: CHANNEL_ID,
  });
  const hasControlCommand = core.channel.text.hasControlCommand(rawBody, config as OpenClawConfig);
  const shouldRequireMention = isGroup
    ? resolveNextcloudTalkGroupRequireMention({
        cfg: config as OpenClawConfig,
        accountId: account.accountId,
        groupId: roomToken,
      })
    : false;
  const { groupPolicy, providerMissingFallbackApplied } =
    resolveAllowlistProviderRuntimeGroupPolicy({
      providerConfigPresent:
        ((config.channels as Record<string, unknown> | undefined)?.[CHANNEL_ID] ?? undefined) !==
        undefined,
      groupPolicy: account.config.groupPolicy,
      defaultGroupPolicy: resolveDefaultGroupPolicy(config as OpenClawConfig),
    });
  const allowFrom = normalizeStringEntries(account.config.allowFrom);
  const outerGroupAllowFrom = account.config.groupAllowFrom?.length
    ? normalizeStringEntries(account.config.groupAllowFrom)
    : allowFrom;
  const roomAllowFrom = normalizeStringEntries(roomConfig?.allowFrom);
  const resolveAccess = async (
    wasMentioned?: boolean,
    contextBinding?: ChannelIngressContextBinding,
  ) =>
    await resolveStableChannelMessageIngress({
      channelId: CHANNEL_ID,
      accountId: account.accountId,
      identity: {
        key: "nextcloud-talk-user-id",
        normalize: (value) => normalizeNextcloudTalkAllowEntry(value) || null,
        sensitivity: "pii",
        entryIdPrefix: "nextcloud-talk-entry",
      },
      cfg: config as OpenClawConfig,
      readStoreAllowFrom: async () =>
        await pairing.readStoreForDmPolicy(CHANNEL_ID, account.accountId),
      subject: { stableId: senderId },
      conversation: {
        kind: isGroup ? "group" : "direct",
        id: isGroup ? roomToken : senderId,
      },
      contextBinding,
      route: roomRoutes({
        isGroup,
        groupPolicy,
        roomMatch,
        roomConfig,
        senderId,
        outerGroupAllowFrom,
        roomAllowFrom,
      }),
      dmPolicy: account.config.dmPolicy ?? "pairing",
      groupPolicy,
      policy: {
        groupAllowFromFallbackToAllowFrom: true,
        activation: {
          requireMention: isGroup && shouldRequireMention,
          allowTextCommands,
        },
      },
      mentionFacts:
        isGroup && wasMentioned !== undefined
          ? {
              canDetectMention: true,
              wasMentioned,
              hasAnyMention: wasMentioned,
            }
          : undefined,
      allowFrom,
      groupAllowFrom: account.config.groupAllowFrom,
      command: {
        allowTextCommands,
        hasControlCommand,
      },
    });
  let access = await resolveAccess();
  warnMissingProviderGroupPolicyFallbackOnce({
    providerMissingFallbackApplied,
    providerKey: "nextcloud-talk",
    accountId: account.accountId,
    blockedLabel: GROUP_POLICY_BLOCKED_LABEL.room,
    log: (messageValue) => runtime.log?.(messageValue),
  });
  const commandAuthorized = access.commandAccess.authorized;
  const accessReason =
    access.ingress.reasonCode === "route_blocked"
      ? "route blocked"
      : access.senderAccess.reasonCode;

  if (isGroup) {
    if (access.routeAccess.reason === "room_not_allowlisted") {
      runtime.log?.(`nextcloud-talk: drop room ${roomToken} (not allowlisted)`);
      return;
    }
    if (access.routeAccess.reason === "room_disabled") {
      runtime.log?.(`nextcloud-talk: drop room ${roomToken} (disabled)`);
      return;
    }
    if (access.routeAccess.reason === "room_sender_not_allowlisted") {
      runtime.log?.(`nextcloud-talk: drop group sender ${senderId} (policy=${groupPolicy})`);
      return;
    }
    if (access.senderAccess.decision !== "allow") {
      runtime.log?.(`nextcloud-talk: drop group sender ${senderId} (reason=${accessReason})`);
      return;
    }
  } else if (access.senderAccess.decision !== "allow") {
    if (access.senderAccess.decision === "pairing") {
      await pairing.issueChallenge({
        senderId,
        senderIdLine: `Your Nextcloud user id: ${senderId}`,
        meta: { name: senderName || undefined },
        sendPairingReply: async (text) => {
          await sendMessageNextcloudTalk(roomToken, text, {
            cfg: config,
            accountId: account.accountId,
          });
          statusSink?.({ lastOutboundAt: Date.now() });
        },
        onReplyError: (err) => {
          runtime.error?.(`nextcloud-talk: pairing reply failed for ${senderId}: ${String(err)}`);
        },
      });
    }
    runtime.log?.(`nextcloud-talk: drop DM sender ${senderId} (reason=${accessReason})`);
    return;
  }

  if (access.commandAccess.shouldBlockControlCommand) {
    logInboundDrop({
      log: (messageLocal) => runtime.log?.(messageLocal),
      channel: CHANNEL_ID,
      reason: "control command (unauthorized)",
      target: senderId,
    });
    return;
  }

  const mentionRegexes = core.channel.mentions.buildMentionRegexes(config as OpenClawConfig);
  const wasMentioned = mentionRegexes.length
    ? core.channel.mentions.matchesMentionPatterns(rawBody, mentionRegexes)
    : false;
  const { route, buildEnvelope } = resolveChannelInboundRouteEnvelope({
    cfg: config as OpenClawConfig,
    channel: CHANNEL_ID,
    accountId: account.accountId,
    peer: {
      kind: isGroup ? "group" : "direct",
      id: isGroup ? roomToken : senderId,
    },
  });
  access = await resolveAccess(isGroup ? wasMentioned : undefined, {
    agentId: route.agentId,
    sessionKey: route.sessionKey,
    messageId: message.messageId,
    inboundEventKind: "user_request",
  });

  if (access.ingress.admission !== "dispatch") {
    runtime.log?.(
      isGroup && access.activationAccess.shouldSkip
        ? `nextcloud-talk: drop room ${roomToken} (no mention)`
        : `nextcloud-talk: drop ${isGroup ? "room" : "DM"} ${roomToken} (authorization changed)`,
    );
    return;
  }

  const fromLabel = isGroup ? `room:${roomName || roomToken}` : senderName || `user:${senderId}`;
  const body = buildEnvelope({
    channel: "Nextcloud Talk",
    from: fromLabel,
    timestamp: message.timestamp,
    body: rawBody,
  });

  const groupSystemPrompt = normalizeOptionalString(roomConfig?.systemPrompt);
  const blockStreamingEnabled = resolveChannelStreamingBlockEnabled(account.config);

  const ctxPayload = core.channel.inbound.buildContext({
    channelIngress: access,
    channel: CHANNEL_ID,
    accountId: route.accountId,
    messageId: message.messageId,
    timestamp: message.timestamp,
    from: isGroup ? `nextcloud-talk:room:${roomToken}` : `nextcloud-talk:${senderId}`,
    sender: { id: senderId, name: senderName || undefined },
    conversation: {
      kind: isGroup ? "group" : "direct",
      id: isGroup ? roomToken : senderId,
      label: fromLabel,
    },
    route: {
      agentId: route.agentId,
      dmScope: route.dmScope,
      accountId: route.accountId,
      routeSessionKey: route.sessionKey,
    },
    reply: { to: `nextcloud-talk:${roomToken}`, originatingTo: `nextcloud-talk:${roomToken}` },
    message: { body, bodyForAgent: rawBody, rawBody, commandBody: rawBody },
    access: {
      commands: { authorized: commandAuthorized },
      mentions: { canDetectMention: isGroup, wasMentioned: isGroup && wasMentioned },
    },
    extra: {
      GroupSubject: isGroup ? roomName || roomToken : undefined,
      GroupSystemPrompt: isGroup ? groupSystemPrompt : undefined,
    },
  });

  await core.channel.inbound.dispatch({
    cfg: config as OpenClawConfig,
    channel: CHANNEL_ID,
    accountId: account.accountId,
    route: { agentId: route.agentId, sessionKey: route.sessionKey },
    ctxPayload,
    delivery: {
      preparePayload: (payload) =>
        payload.text === undefined
          ? payload
          : {
              ...payload,
              text: sanitizeAssistantVisibleText(payload.text),
            },
      deliver: async (payload) => {
        return await deliverNextcloudTalkReply({
          cfg: config,
          payload,
          roomToken,
          accountId: account.accountId,
          statusSink,
        });
      },
      onError: (err, info) => {
        runtime.error?.(`nextcloud-talk ${info.kind} reply failed: ${String(err)}`);
      },
    },
    replyPipeline: {},
    replyOptions: {
      ...(params.turnAdoptionLifecycle
        ? bindIngressLifecycleToReplyOptions(params.turnAdoptionLifecycle)
        : {}),
      skillFilter: roomConfig?.skills,
      disableBlockStreaming:
        typeof blockStreamingEnabled === "boolean" ? !blockStreamingEnabled : undefined,
    },
    record: {
      onRecordError: (err) => {
        runtime.error?.(`nextcloud-talk: failed updating session meta: ${String(err)}`);
      },
    },
  });
}