// Device Pair plugin module implements notify behavior. import { randomUUID } from "node:crypto"; import type { OpenClawPluginService } from "openclaw/plugin-sdk/core"; import { listDevicePairing } from "openclaw/plugin-sdk/device-bootstrap"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; import type { PluginStateKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime"; import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; import { DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS, DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES, DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE, DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, notifyRequestStoreKey, notifySubscriberKey, notifySubscriberStoreKey, type NotifySeenRequest, type NotifySubscription, } from "./notify-state.js"; const NOTIFY_POLL_INTERVAL_MS = 10_000; // Config reload recreates plugin services before an uncancellable delivery may settle. // Keep one module-owned poll so the replacement service cannot race state or delivery. let notifyPollInFlight: Promise | null = null; type PendingPairingRequest = { requestId: string; deviceId: string; displayName?: string; platform?: string; role?: string; roles?: string[]; scopes?: string[]; remoteIp?: string; ts?: number; }; function formatStringList(values?: readonly string[]): string { if (!Array.isArray(values) || values.length === 0) { return "none"; } const normalized = values.map((value) => value.trim()).filter((value) => value.length > 0); return normalized.length > 0 ? normalized.join(", ") : "none"; } function formatRoleList(request: PendingPairingRequest): string { const role = normalizeOptionalString(request.role); if (role) { return role; } return formatStringList(request.roles); } function formatScopeList(request: PendingPairingRequest): string { return formatStringList(request.scopes); } export function formatPendingRequests(pending: PendingPairingRequest[]): string { if (pending.length === 0) { return "No pending device pairing requests."; } const lines: string[] = ["Pending device pairing requests:"]; for (const req of pending) { const label = normalizeOptionalString(req.displayName) || req.deviceId; const platform = normalizeOptionalString(req.platform); const ip = normalizeOptionalString(req.remoteIp); const parts = [ `- ${req.requestId}`, label ? `name=${label}` : null, platform ? `platform=${platform}` : null, `role=${formatRoleList(req)}`, `scopes=${formatScopeList(req)}`, ip ? `ip=${ip}` : null, ].filter(Boolean); lines.push(parts.join(" ยท ")); } return lines.join("\n"); } type NotifySubscriberStore = PluginStateKeyedStore & { deleteIf: NonNullable["deleteIf"]>; }; function openNotifySubscriberStore(api: OpenClawPluginApi): NotifySubscriberStore { const store = api.runtime.state.openKeyedStore({ namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, }); if (!store.deleteIf) { throw new Error( "device-pair notify requires a runtime with atomic plugin state conditional delete support", ); } return store as NotifySubscriberStore; } function openNotifySeenRequestStore( api: OpenClawPluginApi, ): PluginStateKeyedStore { return api.runtime.state.openKeyedStore({ namespace: DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE, maxEntries: DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES, defaultTtlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS, }); } type NotifyTarget = { to: string; accountId?: string; messageThreadId?: string | number; }; function resolveNotifyTarget(ctx: { senderId?: string; from?: string; to?: string; accountId?: string; messageThreadId?: string | number; }): NotifyTarget | null { const to = normalizeOptionalString(ctx.senderId) || normalizeOptionalString(ctx.from) || normalizeOptionalString(ctx.to) || ""; if (!to) { return null; } return { to, ...(ctx.accountId ? { accountId: ctx.accountId } : {}), ...(ctx.messageThreadId != null ? { messageThreadId: ctx.messageThreadId } : {}), }; } function nextNotifySubscription( target: NotifyTarget, mode: NotifySubscription["mode"], ): NotifySubscription { return { ...target, mode, addedAtMs: Date.now(), armId: randomUUID(), }; } async function registerNotifySubscriber(params: { api: OpenClawPluginApi; target: NotifyTarget; mode: NotifySubscription["mode"]; refresh: boolean; }): Promise { const store = openNotifySubscriberStore(params.api); const key = notifySubscriberStoreKey(params.target); const current = await store.lookup(key); if (!params.refresh && current?.mode === params.mode) { return false; } await store.register(key, nextNotifySubscription(params.target, params.mode)); return true; } function isSameNotifySubscription( current: NotifySubscription, expected: NotifySubscription, ): boolean { if (expected.armId) { return current.armId === expected.armId; } // Doctor-imported legacy subscriptions have no arm id. Their original // fields remain the exact generation until a new arm replaces the row. return ( current.armId === undefined && current.mode === expected.mode && current.addedAtMs === expected.addedAtMs && notifySubscriberKey(current) === notifySubscriberKey(expected) ); } function buildPairingRequestNotificationText(request: PendingPairingRequest): string { const label = normalizeOptionalString(request.displayName) || request.deviceId; const platform = normalizeOptionalString(request.platform); const ip = normalizeOptionalString(request.remoteIp); const role = formatRoleList(request); const scopes = formatScopeList(request); const lines = [ "๐Ÿ“ฒ New device pairing request", `ID: ${request.requestId}`, `Name: ${label}`, ...(platform ? [`Platform: ${platform}`] : []), `Role: ${role}`, `Scopes: ${scopes}`, ...(ip ? [`IP: ${ip}`] : []), "", `Approve: /pair approve ${request.requestId}`, "List pending: /pair pending", ]; return lines.join("\n"); } function requestTimestampMs(request: PendingPairingRequest): number | null { if (typeof request.ts !== "number" || !Number.isFinite(request.ts)) { return null; } const ts = Math.trunc(request.ts); return ts > 0 ? ts : null; } function shouldNotifySubscriberForRequest( subscriber: NotifySubscription, request: PendingPairingRequest, ): boolean { if (subscriber.mode !== "once") { return true; } const ts = requestTimestampMs(request); // One-shot subscriptions should only notify for new requests created after arming. if (ts == null) { return false; } return ts >= subscriber.addedAtMs; } async function notifySubscriber(params: { api: OpenClawPluginApi; subscriber: NotifySubscription; text: string; }): Promise { const adapter = await params.api.runtime.channel.outbound.loadAdapter("telegram"); const send = adapter?.sendText; if (!send) { params.api.logger.warn( "device-pair: telegram outbound adapter unavailable for pairing notifications", ); return false; } try { await send({ cfg: params.api.config, to: params.subscriber.to, text: params.text, ...(params.subscriber.accountId ? { accountId: params.subscriber.accountId } : {}), ...(params.subscriber.messageThreadId != null ? { threadId: params.subscriber.messageThreadId } : {}), }); return true; } catch (err) { params.api.logger.warn( `device-pair: failed to send pairing notification to ${params.subscriber.to}: ${formatErrorMessage(err)}`, ); return false; } } async function notifyPendingPairingRequests(params: { api: OpenClawPluginApi }): Promise { const subscriberStore = openNotifySubscriberStore(params.api); const seenRequestStore = openNotifySeenRequestStore(params.api); const [subscriberEntries, seenRequestEntries, pairing] = await Promise.all([ subscriberStore.entries(), seenRequestStore.entries(), listDevicePairing(), ]); const subscribers = subscriberEntries.toSorted((a, b) => a.value.addedAtMs - b.value.addedAtMs); const pending: PendingPairingRequest[] = pairing.pending; const now = Date.now(); const pendingIds = new Set(pending.map((entry) => entry.requestId)); const notifiedRequestIds = new Set(); for (const entry of seenRequestEntries) { const requestId = normalizeOptionalString(entry.value.requestId); const notifiedAtMs = entry.value.notifiedAtMs; if ( !requestId || !Number.isFinite(notifiedAtMs) || notifiedAtMs <= 0 || !pendingIds.has(requestId) || now - notifiedAtMs > DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS ) { await seenRequestStore.delete(entry.key); continue; } notifiedRequestIds.add(requestId); } if (subscribers.length > 0) { const deliveredOneShots = new Set(); for (const request of pending) { if (notifiedRequestIds.has(request.requestId)) { continue; } const text = buildPairingRequestNotificationText(request); let delivered = false; for (const entry of subscribers) { const subscriber = entry.value; if (subscriber.mode === "once" && deliveredOneShots.has(entry.key)) { continue; } if (!shouldNotifySubscriberForRequest(subscriber, request)) { continue; } const sent = await notifySubscriber({ api: params.api, subscriber, text, }); delivered = delivered || sent; if (sent && subscriber.mode === "once") { deliveredOneShots.add(entry.key); // Delivery is fallible and uncancellable. Delete only the exact arm // that was sent so an overlapping re-arm remains subscribed. await subscriberStore.deleteIf(entry.key, (current) => isSameNotifySubscription(current, subscriber), ); } } if (delivered) { await seenRequestStore.register( notifyRequestStoreKey(request.requestId), { requestId: request.requestId, notifiedAtMs: now }, { ttlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS }, ); notifiedRequestIds.add(request.requestId); } } } } async function runNotifyPoll(api: OpenClawPluginApi): Promise { if (notifyPollInFlight) { return; } notifyPollInFlight = notifyPendingPairingRequests({ api }); try { await notifyPollInFlight; } finally { notifyPollInFlight = null; } } export async function armPairNotifyOnce(params: { api: OpenClawPluginApi; ctx: { channel: string; senderId?: string; from?: string; to?: string; accountId?: string; messageThreadId?: string | number; }; }): Promise { if (params.ctx.channel !== "telegram") { return false; } const target = resolveNotifyTarget(params.ctx); if (!target) { return false; } await registerNotifySubscriber({ api: params.api, target, mode: "once", refresh: true, }); return true; } export async function handleNotifyCommand(params: { api: OpenClawPluginApi; ctx: { channel: string; senderId?: string; from?: string; to?: string; accountId?: string; messageThreadId?: string | number; }; action: string; }): Promise<{ text: string }> { if (params.ctx.channel !== "telegram") { return { text: "Pairing notifications are currently supported only on Telegram." }; } const target = resolveNotifyTarget(params.ctx); if (!target) { return { text: "Could not resolve Telegram target for this chat." }; } const subscriberStore = openNotifySubscriberStore(params.api); const targetStoreKey = notifySubscriberStoreKey(target); if (params.action === "on" || params.action === "enable") { await registerNotifySubscriber({ api: params.api, target, mode: "persistent", refresh: false, }); return { text: "โœ… Pair request notifications enabled for this Telegram chat.\n" + "I will ping here when a new device pairing request arrives.", }; } if (params.action === "off" || params.action === "disable") { await subscriberStore.delete(targetStoreKey); return { text: "โœ… Pair request notifications disabled for this Telegram chat." }; } if (params.action === "once" || params.action === "arm") { await armPairNotifyOnce({ api: params.api, ctx: params.ctx, }); return { text: "โœ… One-shot pairing notification armed for this Telegram chat.\n" + "I will notify on the next new pairing request, then auto-disable.", }; } if (params.action === "status" || params.action === "") { const [current, subscribers, pending] = await Promise.all([ subscriberStore.lookup(targetStoreKey), subscriberStore.entries(), listDevicePairing(), ]); const enabled = Boolean(current); const mode = current?.mode ?? "off"; return { text: [ `Pair request notifications: ${enabled ? "enabled" : "disabled"} for this chat.`, `Mode: ${mode}`, `Subscribers: ${subscribers.length}`, `Pending requests: ${pending.pending.length}`, "", "Use /pair notify on|off|once", ].join("\n"), }; } return { text: "Usage: /pair notify on|off|once|status" }; } export function createPairingNotifierService(api: OpenClawPluginApi): OpenClawPluginService { let notifyInterval: ReturnType | null = null; return { id: "device-pair-notifier", start: () => { const tick = async () => { await runNotifyPoll(api); }; // Pairing notifications are eventual background work. Starting on the // existing interval keeps SQLite pairing scans out of Gateway readiness. notifyInterval = setInterval(() => { tick().catch((err: unknown) => { api.logger.warn(`device-pair: notify poll failed: ${formatErrorMessage(err)}`); }); }, NOTIFY_POLL_INTERVAL_MS); notifyInterval.unref?.(); }, stop: async () => { if (notifyInterval) { clearInterval(notifyInterval); notifyInterval = null; } }, }; }