mirror of
https://github.com/moltbot/moltbot.git
synced 2026-03-08 06:54:24 +00:00
Fix BlueBubbles DM history backfill bug (#20302)
* feat: implement DM history backfill for BlueBubbles - Add fetchBlueBubblesHistory function to fetch message history from API - Modify processMessage to fetch history for both groups and DMs - Use dmHistoryLimit for DMs and historyLimit for groups - Add InboundHistory field to finalizeInboundContext call Fixes #20296 * style: format with oxfmt * address review: in-memory history cache, resolveAccount try/catch, include is_from_me - Wrap resolveAccount in try/catch instead of unreachable guard (it throws) - Include is_from_me messages with 'me' sender label for full conversation context - Add in-memory rolling history map (chatHistories) matching other channel patterns - API backfill only on first message per chat, not every incoming message - Remove unused buildInboundHistoryFromEntries import * chore: remove unused buildInboundHistoryFromEntries helper Dead code flagged by Greptile — mapping is done inline in monitor-processing.ts. * BlueBubbles: harden DM history backfill state handling * BlueBubbles: add bounded exponential backoff and history payload guards * BlueBubbles: evict merged history keys * Update extensions/bluebubbles/src/monitor-processing.ts Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --------- Co-authored-by: Ryan Mac Mini <ryanmacmini@ryans-mac-mini.tailf78f8b.ts.net> Co-authored-by: Vincent Koc <vincentkoc@ieee.org> Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
This commit is contained in:
Ryan Haines
177
extensions/bluebubbles/src/history.ts
Normal file
177
extensions/bluebubbles/src/history.ts
Normal file
@@ -0,0 +1,177 @@
|
||||
import type { OpenClawConfig } from "openclaw/plugin-sdk";
|
||||
import { resolveBlueBubblesServerAccount } from "./account-resolve.js";
|
||||
import { blueBubblesFetchWithTimeout, buildBlueBubblesApiUrl } from "./types.js";
|
||||
|
||||
export type BlueBubblesHistoryEntry = {
|
||||
sender: string;
|
||||
body: string;
|
||||
timestamp?: number;
|
||||
messageId?: string;
|
||||
};
|
||||
|
||||
export type BlueBubblesHistoryFetchResult = {
|
||||
entries: BlueBubblesHistoryEntry[];
|
||||
/**
|
||||
* True when at least one API path returned a recognized response shape.
|
||||
* False means all attempts failed or returned unusable data.
|
||||
*/
|
||||
resolved: boolean;
|
||||
};
|
||||
|
||||
export type BlueBubblesMessageData = {
|
||||
guid?: string;
|
||||
text?: string;
|
||||
handle_id?: string;
|
||||
is_from_me?: boolean;
|
||||
date_created?: number;
|
||||
date_delivered?: number;
|
||||
associated_message_guid?: string;
|
||||
sender?: {
|
||||
address?: string;
|
||||
display_name?: string;
|
||||
};
|
||||
};
|
||||
|
||||
export type BlueBubblesChatOpts = {
|
||||
serverUrl?: string;
|
||||
password?: string;
|
||||
accountId?: string;
|
||||
timeoutMs?: number;
|
||||
cfg?: OpenClawConfig;
|
||||
};
|
||||
|
||||
function resolveAccount(params: BlueBubblesChatOpts) {
|
||||
return resolveBlueBubblesServerAccount(params);
|
||||
}
|
||||
|
||||
const MAX_HISTORY_FETCH_LIMIT = 100;
|
||||
const HISTORY_SCAN_MULTIPLIER = 8;
|
||||
const MAX_HISTORY_SCAN_MESSAGES = 500;
|
||||
const MAX_HISTORY_BODY_CHARS = 2_000;
|
||||
|
||||
function clampHistoryLimit(limit: number): number {
|
||||
if (!Number.isFinite(limit)) {
|
||||
return 0;
|
||||
}
|
||||
const normalized = Math.floor(limit);
|
||||
if (normalized <= 0) {
|
||||
return 0;
|
||||
}
|
||||
return Math.min(normalized, MAX_HISTORY_FETCH_LIMIT);
|
||||
}
|
||||
|
||||
function truncateHistoryBody(text: string): string {
|
||||
if (text.length <= MAX_HISTORY_BODY_CHARS) {
|
||||
return text;
|
||||
}
|
||||
return `${text.slice(0, MAX_HISTORY_BODY_CHARS).trimEnd()}...`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch message history from BlueBubbles API for a specific chat.
|
||||
* This provides the initial backfill for both group chats and DMs.
|
||||
*/
|
||||
export async function fetchBlueBubblesHistory(
|
||||
chatIdentifier: string,
|
||||
limit: number,
|
||||
opts: BlueBubblesChatOpts = {},
|
||||
): Promise<BlueBubblesHistoryFetchResult> {
|
||||
const effectiveLimit = clampHistoryLimit(limit);
|
||||
if (!chatIdentifier.trim() || effectiveLimit <= 0) {
|
||||
return { entries: [], resolved: true };
|
||||
}
|
||||
|
||||
let baseUrl: string;
|
||||
let password: string;
|
||||
try {
|
||||
({ baseUrl, password } = resolveAccount(opts));
|
||||
} catch {
|
||||
return { entries: [], resolved: false };
|
||||
}
|
||||
|
||||
// Try different common API patterns for fetching messages
|
||||
const possiblePaths = [
|
||||
`/api/v1/chat/${encodeURIComponent(chatIdentifier)}/messages?limit=${effectiveLimit}&sort=DESC`,
|
||||
`/api/v1/messages?chatGuid=${encodeURIComponent(chatIdentifier)}&limit=${effectiveLimit}`,
|
||||
`/api/v1/chat/${encodeURIComponent(chatIdentifier)}/message?limit=${effectiveLimit}`,
|
||||
];
|
||||
|
||||
for (const path of possiblePaths) {
|
||||
try {
|
||||
const url = buildBlueBubblesApiUrl({ baseUrl, path, password });
|
||||
const res = await blueBubblesFetchWithTimeout(
|
||||
url,
|
||||
{ method: "GET" },
|
||||
opts.timeoutMs ?? 10000,
|
||||
);
|
||||
|
||||
if (!res.ok) {
|
||||
continue; // Try next path
|
||||
}
|
||||
|
||||
const data = await res.json().catch(() => null);
|
||||
if (!data) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Handle different response structures
|
||||
let messages: unknown[] = [];
|
||||
if (Array.isArray(data)) {
|
||||
messages = data;
|
||||
} else if (data.data && Array.isArray(data.data)) {
|
||||
messages = data.data;
|
||||
} else if (data.messages && Array.isArray(data.messages)) {
|
||||
messages = data.messages;
|
||||
} else {
|
||||
continue;
|
||||
}
|
||||
|
||||
const historyEntries: BlueBubblesHistoryEntry[] = [];
|
||||
|
||||
const maxScannedMessages = Math.min(
|
||||
Math.max(effectiveLimit * HISTORY_SCAN_MULTIPLIER, effectiveLimit),
|
||||
MAX_HISTORY_SCAN_MESSAGES,
|
||||
);
|
||||
for (let i = 0; i < messages.length && i < maxScannedMessages; i++) {
|
||||
const item = messages[i];
|
||||
const msg = item as BlueBubblesMessageData;
|
||||
|
||||
// Skip messages without text content
|
||||
const text = msg.text?.trim();
|
||||
if (!text) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const sender = msg.is_from_me
|
||||
? "me"
|
||||
: msg.sender?.display_name || msg.sender?.address || msg.handle_id || "Unknown";
|
||||
const timestamp = msg.date_created || msg.date_delivered;
|
||||
|
||||
historyEntries.push({
|
||||
sender,
|
||||
body: truncateHistoryBody(text),
|
||||
timestamp,
|
||||
messageId: msg.guid,
|
||||
});
|
||||
}
|
||||
|
||||
// Sort by timestamp (oldest first for context)
|
||||
historyEntries.sort((a, b) => {
|
||||
const aTime = a.timestamp || 0;
|
||||
const bTime = b.timestamp || 0;
|
||||
return aTime - bTime;
|
||||
});
|
||||
|
||||
return {
|
||||
entries: historyEntries.slice(0, effectiveLimit), // Ensure we don't exceed the requested limit
|
||||
resolved: true,
|
||||
};
|
||||
} catch (error) {
|
||||
// Continue to next path
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// If none of the API paths worked, return empty history
|
||||
return { entries: [], resolved: false };
|
||||
}
|
||||
@@ -1,17 +1,21 @@
|
||||
import type { OpenClawConfig } from "openclaw/plugin-sdk";
|
||||
import {
|
||||
createReplyPrefixOptions,
|
||||
evictOldHistoryKeys,
|
||||
logAckFailure,
|
||||
logInboundDrop,
|
||||
logTypingFailure,
|
||||
recordPendingHistoryEntryIfEnabled,
|
||||
resolveAckReaction,
|
||||
resolveDmGroupAccessDecision,
|
||||
resolveEffectiveAllowFromLists,
|
||||
resolveControlCommandGate,
|
||||
stripMarkdown,
|
||||
type HistoryEntry,
|
||||
} from "openclaw/plugin-sdk";
|
||||
import { downloadBlueBubblesAttachment } from "./attachments.js";
|
||||
import { markBlueBubblesChatRead, sendBlueBubblesTyping } from "./chat.js";
|
||||
import { fetchBlueBubblesHistory } from "./history.js";
|
||||
import { sendBlueBubblesMedia } from "./media-send.js";
|
||||
import {
|
||||
buildMessagePlaceholder,
|
||||
@@ -239,6 +243,178 @@ function resolveBlueBubblesAckReaction(params: {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory rolling history map keyed by account + chat identifier.
|
||||
* Populated from incoming messages during the session.
|
||||
* API backfill is attempted until one fetch resolves (or retries are exhausted).
|
||||
*/
|
||||
const chatHistories = new Map<string, HistoryEntry[]>();
|
||||
type HistoryBackfillState = {
|
||||
attempts: number;
|
||||
firstAttemptAt: number;
|
||||
nextAttemptAt: number;
|
||||
resolved: boolean;
|
||||
};
|
||||
|
||||
const historyBackfills = new Map<string, HistoryBackfillState>();
|
||||
const HISTORY_BACKFILL_BASE_DELAY_MS = 5_000;
|
||||
const HISTORY_BACKFILL_MAX_DELAY_MS = 2 * 60 * 1000;
|
||||
const HISTORY_BACKFILL_MAX_ATTEMPTS = 6;
|
||||
const HISTORY_BACKFILL_RETRY_WINDOW_MS = 30 * 60 * 1000;
|
||||
const MAX_STORED_HISTORY_ENTRY_CHARS = 2_000;
|
||||
const MAX_INBOUND_HISTORY_ENTRY_CHARS = 1_200;
|
||||
const MAX_INBOUND_HISTORY_TOTAL_CHARS = 12_000;
|
||||
|
||||
function buildAccountScopedHistoryKey(accountId: string, historyIdentifier: string): string {
|
||||
return `${accountId}\u0000${historyIdentifier}`;
|
||||
}
|
||||
|
||||
function historyDedupKey(entry: HistoryEntry): string {
|
||||
const messageId = entry.messageId?.trim();
|
||||
if (messageId) {
|
||||
return `id:${messageId}`;
|
||||
}
|
||||
return `fallback:${entry.sender}\u0000${entry.body}\u0000${entry.timestamp ?? ""}`;
|
||||
}
|
||||
|
||||
function truncateHistoryBody(body: string, maxChars: number): string {
|
||||
const trimmed = body.trim();
|
||||
if (!trimmed) {
|
||||
return "";
|
||||
}
|
||||
if (trimmed.length <= maxChars) {
|
||||
return trimmed;
|
||||
}
|
||||
return `${trimmed.slice(0, maxChars).trimEnd()}...`;
|
||||
}
|
||||
|
||||
function mergeHistoryEntries(params: {
|
||||
apiEntries: HistoryEntry[];
|
||||
currentEntries: HistoryEntry[];
|
||||
limit: number;
|
||||
}): HistoryEntry[] {
|
||||
if (params.limit <= 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const merged: HistoryEntry[] = [];
|
||||
const seen = new Set<string>();
|
||||
const appendUnique = (entry: HistoryEntry) => {
|
||||
const key = historyDedupKey(entry);
|
||||
if (seen.has(key)) {
|
||||
return;
|
||||
}
|
||||
seen.add(key);
|
||||
merged.push(entry);
|
||||
};
|
||||
|
||||
for (const entry of params.apiEntries) {
|
||||
appendUnique(entry);
|
||||
}
|
||||
for (const entry of params.currentEntries) {
|
||||
appendUnique(entry);
|
||||
}
|
||||
|
||||
if (merged.length <= params.limit) {
|
||||
return merged;
|
||||
}
|
||||
return merged.slice(merged.length - params.limit);
|
||||
}
|
||||
|
||||
function pruneHistoryBackfillState(): void {
|
||||
for (const key of historyBackfills.keys()) {
|
||||
if (!chatHistories.has(key)) {
|
||||
historyBackfills.delete(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function markHistoryBackfillResolved(historyKey: string): void {
|
||||
const state = historyBackfills.get(historyKey);
|
||||
if (state) {
|
||||
state.resolved = true;
|
||||
historyBackfills.set(historyKey, state);
|
||||
return;
|
||||
}
|
||||
historyBackfills.set(historyKey, {
|
||||
attempts: 0,
|
||||
firstAttemptAt: Date.now(),
|
||||
nextAttemptAt: Number.POSITIVE_INFINITY,
|
||||
resolved: true,
|
||||
});
|
||||
}
|
||||
|
||||
function planHistoryBackfillAttempt(historyKey: string, now: number): HistoryBackfillState | null {
|
||||
const existing = historyBackfills.get(historyKey);
|
||||
if (existing?.resolved) {
|
||||
return null;
|
||||
}
|
||||
if (existing && now - existing.firstAttemptAt > HISTORY_BACKFILL_RETRY_WINDOW_MS) {
|
||||
markHistoryBackfillResolved(historyKey);
|
||||
return null;
|
||||
}
|
||||
if (existing && existing.attempts >= HISTORY_BACKFILL_MAX_ATTEMPTS) {
|
||||
markHistoryBackfillResolved(historyKey);
|
||||
return null;
|
||||
}
|
||||
if (existing && now < existing.nextAttemptAt) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const attempts = (existing?.attempts ?? 0) + 1;
|
||||
const firstAttemptAt = existing?.firstAttemptAt ?? now;
|
||||
const backoffDelay = Math.min(
|
||||
HISTORY_BACKFILL_BASE_DELAY_MS * 2 ** (attempts - 1),
|
||||
HISTORY_BACKFILL_MAX_DELAY_MS,
|
||||
);
|
||||
const state: HistoryBackfillState = {
|
||||
attempts,
|
||||
firstAttemptAt,
|
||||
nextAttemptAt: now + backoffDelay,
|
||||
resolved: false,
|
||||
};
|
||||
historyBackfills.set(historyKey, state);
|
||||
return state;
|
||||
}
|
||||
|
||||
function buildInboundHistorySnapshot(params: {
|
||||
entries: HistoryEntry[];
|
||||
limit: number;
|
||||
}): Array<{ sender: string; body: string; timestamp?: number }> | undefined {
|
||||
if (params.limit <= 0 || params.entries.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
const recent = params.entries.slice(-params.limit);
|
||||
const selected: Array<{ sender: string; body: string; timestamp?: number }> = [];
|
||||
let remainingChars = MAX_INBOUND_HISTORY_TOTAL_CHARS;
|
||||
|
||||
for (let i = recent.length - 1; i >= 0; i--) {
|
||||
const entry = recent[i];
|
||||
const body = truncateHistoryBody(entry.body, MAX_INBOUND_HISTORY_ENTRY_CHARS);
|
||||
if (!body) {
|
||||
continue;
|
||||
}
|
||||
if (selected.length > 0 && body.length > remainingChars) {
|
||||
break;
|
||||
}
|
||||
selected.push({
|
||||
sender: entry.sender,
|
||||
body,
|
||||
timestamp: entry.timestamp,
|
||||
});
|
||||
remainingChars -= body.length;
|
||||
if (remainingChars <= 0) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (selected.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
selected.reverse();
|
||||
return selected;
|
||||
}
|
||||
|
||||
export async function processMessage(
|
||||
message: NormalizedWebhookMessage,
|
||||
target: WebhookTarget,
|
||||
@@ -808,9 +984,118 @@ export async function processMessage(
|
||||
.trim();
|
||||
};
|
||||
|
||||
// History: in-memory rolling map with bounded API backfill retries
|
||||
const historyLimit = isGroup
|
||||
? (account.config.historyLimit ?? 0)
|
||||
: (account.config.dmHistoryLimit ?? 0);
|
||||
|
||||
const historyIdentifier =
|
||||
chatGuid ||
|
||||
chatIdentifier ||
|
||||
(chatId ? String(chatId) : null) ||
|
||||
(isGroup ? null : message.senderId) ||
|
||||
"";
|
||||
const historyKey = historyIdentifier
|
||||
? buildAccountScopedHistoryKey(account.accountId, historyIdentifier)
|
||||
: "";
|
||||
|
||||
// Record the current message into rolling history
|
||||
if (historyKey && historyLimit > 0) {
|
||||
const nowMs = Date.now();
|
||||
const senderLabel = message.fromMe ? "me" : message.senderName || message.senderId;
|
||||
const normalizedHistoryBody = truncateHistoryBody(text, MAX_STORED_HISTORY_ENTRY_CHARS);
|
||||
const currentEntries = recordPendingHistoryEntryIfEnabled({
|
||||
historyMap: chatHistories,
|
||||
limit: historyLimit,
|
||||
historyKey,
|
||||
entry: normalizedHistoryBody
|
||||
? {
|
||||
sender: senderLabel,
|
||||
body: normalizedHistoryBody,
|
||||
timestamp: message.timestamp ?? nowMs,
|
||||
messageId: message.messageId ?? undefined,
|
||||
}
|
||||
: null,
|
||||
});
|
||||
pruneHistoryBackfillState();
|
||||
|
||||
const backfillAttempt = planHistoryBackfillAttempt(historyKey, nowMs);
|
||||
if (backfillAttempt) {
|
||||
try {
|
||||
const backfillResult = await fetchBlueBubblesHistory(historyIdentifier, historyLimit, {
|
||||
cfg: config,
|
||||
accountId: account.accountId,
|
||||
});
|
||||
if (backfillResult.resolved) {
|
||||
markHistoryBackfillResolved(historyKey);
|
||||
}
|
||||
if (backfillResult.entries.length > 0) {
|
||||
const apiEntries: HistoryEntry[] = [];
|
||||
for (const entry of backfillResult.entries) {
|
||||
const body = truncateHistoryBody(entry.body, MAX_STORED_HISTORY_ENTRY_CHARS);
|
||||
if (!body) {
|
||||
continue;
|
||||
}
|
||||
apiEntries.push({
|
||||
sender: entry.sender,
|
||||
body,
|
||||
timestamp: entry.timestamp,
|
||||
messageId: entry.messageId,
|
||||
});
|
||||
}
|
||||
const merged = mergeHistoryEntries({
|
||||
apiEntries,
|
||||
currentEntries:
|
||||
currentEntries.length > 0 ? currentEntries : (chatHistories.get(historyKey) ?? []),
|
||||
limit: historyLimit,
|
||||
});
|
||||
if (chatHistories.has(historyKey)) {
|
||||
chatHistories.delete(historyKey);
|
||||
}
|
||||
chatHistories.set(historyKey, merged);
|
||||
evictOldHistoryKeys(chatHistories);
|
||||
logVerbose(
|
||||
core,
|
||||
runtime,
|
||||
`backfilled ${backfillResult.entries.length} history messages for ${isGroup ? "group" : "DM"}: ${historyIdentifier}`,
|
||||
);
|
||||
} else if (!backfillResult.resolved) {
|
||||
const remainingAttempts = HISTORY_BACKFILL_MAX_ATTEMPTS - backfillAttempt.attempts;
|
||||
const nextBackoffMs = Math.max(backfillAttempt.nextAttemptAt - nowMs, 0);
|
||||
logVerbose(
|
||||
core,
|
||||
runtime,
|
||||
`history backfill unresolved for ${historyIdentifier}; retries left=${Math.max(remainingAttempts, 0)} next_in_ms=${nextBackoffMs}`,
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
const remainingAttempts = HISTORY_BACKFILL_MAX_ATTEMPTS - backfillAttempt.attempts;
|
||||
const nextBackoffMs = Math.max(backfillAttempt.nextAttemptAt - nowMs, 0);
|
||||
logVerbose(
|
||||
core,
|
||||
runtime,
|
||||
`history backfill failed for ${historyIdentifier}: ${String(err)} (retries left=${Math.max(remainingAttempts, 0)} next_in_ms=${nextBackoffMs})`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Build inbound history from the in-memory map
|
||||
let inboundHistory: Array<{ sender: string; body: string; timestamp?: number }> | undefined;
|
||||
if (historyKey && historyLimit > 0) {
|
||||
const entries = chatHistories.get(historyKey);
|
||||
if (entries && entries.length > 0) {
|
||||
inboundHistory = buildInboundHistorySnapshot({
|
||||
entries,
|
||||
limit: historyLimit,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const ctxPayload = core.channel.reply.finalizeInboundContext({
|
||||
Body: body,
|
||||
BodyForAgent: rawBody,
|
||||
InboundHistory: inboundHistory,
|
||||
RawBody: rawBody,
|
||||
CommandBody: rawBody,
|
||||
BodyForCommands: rawBody,
|
||||
|
||||
@@ -4,6 +4,7 @@ import type { OpenClawConfig, PluginRuntime } from "openclaw/plugin-sdk";
|
||||
import { removeAckReactionAfterReply, shouldAckReaction } from "openclaw/plugin-sdk";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { ResolvedBlueBubblesAccount } from "./accounts.js";
|
||||
import { fetchBlueBubblesHistory } from "./history.js";
|
||||
import {
|
||||
handleBlueBubblesWebhookRequest,
|
||||
registerBlueBubblesWebhookTarget,
|
||||
@@ -38,6 +39,10 @@ vi.mock("./reactions.js", async () => {
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("./history.js", () => ({
|
||||
fetchBlueBubblesHistory: vi.fn().mockResolvedValue({ entries: [], resolved: true }),
|
||||
}));
|
||||
|
||||
// Mock runtime
|
||||
const mockEnqueueSystemEvent = vi.fn();
|
||||
const mockBuildPairingReply = vi.fn(() => "Pairing code: TESTCODE");
|
||||
@@ -86,6 +91,7 @@ const mockChunkByNewline = vi.fn((text: string) => (text ? [text] : []));
|
||||
const mockChunkTextWithMode = vi.fn((text: string) => (text ? [text] : []));
|
||||
const mockChunkMarkdownTextWithMode = vi.fn((text: string) => (text ? [text] : []));
|
||||
const mockResolveChunkMode = vi.fn(() => "length");
|
||||
const mockFetchBlueBubblesHistory = vi.mocked(fetchBlueBubblesHistory);
|
||||
|
||||
function createMockRuntime(): PluginRuntime {
|
||||
return {
|
||||
@@ -355,6 +361,7 @@ describe("BlueBubbles webhook monitor", () => {
|
||||
vi.clearAllMocks();
|
||||
// Reset short ID state between tests for predictable behavior
|
||||
_resetBlueBubblesShortIdState();
|
||||
mockFetchBlueBubblesHistory.mockResolvedValue({ entries: [], resolved: true });
|
||||
mockReadAllowFromStore.mockResolvedValue([]);
|
||||
mockUpsertPairingRequest.mockResolvedValue({ code: "TESTCODE", created: true });
|
||||
mockResolveRequireMention.mockReturnValue(false);
|
||||
@@ -2991,6 +2998,279 @@ describe("BlueBubbles webhook monitor", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("history backfill", () => {
|
||||
it("scopes in-memory history by account to avoid cross-account leakage", async () => {
|
||||
mockFetchBlueBubblesHistory.mockImplementation(async (_chatIdentifier, _limit, opts) => {
|
||||
if (opts?.accountId === "acc-a") {
|
||||
return {
|
||||
resolved: true,
|
||||
entries: [
|
||||
{ sender: "A", body: "a-history", messageId: "a-history-1", timestamp: 1000 },
|
||||
],
|
||||
};
|
||||
}
|
||||
if (opts?.accountId === "acc-b") {
|
||||
return {
|
||||
resolved: true,
|
||||
entries: [
|
||||
{ sender: "B", body: "b-history", messageId: "b-history-1", timestamp: 1000 },
|
||||
],
|
||||
};
|
||||
}
|
||||
return { resolved: true, entries: [] };
|
||||
});
|
||||
|
||||
const accountA: ResolvedBlueBubblesAccount = {
|
||||
...createMockAccount({ dmHistoryLimit: 3, password: "password-a" }),
|
||||
accountId: "acc-a",
|
||||
};
|
||||
const accountB: ResolvedBlueBubblesAccount = {
|
||||
...createMockAccount({ dmHistoryLimit: 3, password: "password-b" }),
|
||||
accountId: "acc-b",
|
||||
};
|
||||
const config: OpenClawConfig = {};
|
||||
const core = createMockRuntime();
|
||||
setBlueBubblesRuntime(core);
|
||||
|
||||
const unregisterA = registerBlueBubblesWebhookTarget({
|
||||
account: accountA,
|
||||
config,
|
||||
runtime: { log: vi.fn(), error: vi.fn() },
|
||||
core,
|
||||
path: "/bluebubbles-webhook",
|
||||
});
|
||||
const unregisterB = registerBlueBubblesWebhookTarget({
|
||||
account: accountB,
|
||||
config,
|
||||
runtime: { log: vi.fn(), error: vi.fn() },
|
||||
core,
|
||||
path: "/bluebubbles-webhook",
|
||||
});
|
||||
unregister = () => {
|
||||
unregisterA();
|
||||
unregisterB();
|
||||
};
|
||||
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook?password=password-a", {
|
||||
type: "new-message",
|
||||
data: {
|
||||
text: "message for account a",
|
||||
handle: { address: "+15551234567" },
|
||||
isGroup: false,
|
||||
isFromMe: false,
|
||||
guid: "a-msg-1",
|
||||
chatGuid: "iMessage;-;+15551234567",
|
||||
date: Date.now(),
|
||||
},
|
||||
}),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook?password=password-b", {
|
||||
type: "new-message",
|
||||
data: {
|
||||
text: "message for account b",
|
||||
handle: { address: "+15551234567" },
|
||||
isGroup: false,
|
||||
isFromMe: false,
|
||||
guid: "b-msg-1",
|
||||
chatGuid: "iMessage;-;+15551234567",
|
||||
date: Date.now(),
|
||||
},
|
||||
}),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
|
||||
expect(mockDispatchReplyWithBufferedBlockDispatcher).toHaveBeenCalledTimes(2);
|
||||
const firstCall = mockDispatchReplyWithBufferedBlockDispatcher.mock.calls[0]?.[0];
|
||||
const secondCall = mockDispatchReplyWithBufferedBlockDispatcher.mock.calls[1]?.[0];
|
||||
const firstHistory = (firstCall?.ctx.InboundHistory ?? []) as Array<{ body: string }>;
|
||||
const secondHistory = (secondCall?.ctx.InboundHistory ?? []) as Array<{ body: string }>;
|
||||
expect(firstHistory.map((entry) => entry.body)).toContain("a-history");
|
||||
expect(secondHistory.map((entry) => entry.body)).toContain("b-history");
|
||||
expect(secondHistory.map((entry) => entry.body)).not.toContain("a-history");
|
||||
});
|
||||
|
||||
it("dedupes and caps merged history to dmHistoryLimit", async () => {
|
||||
mockFetchBlueBubblesHistory.mockResolvedValueOnce({
|
||||
resolved: true,
|
||||
entries: [
|
||||
{ sender: "Friend", body: "older context", messageId: "hist-1", timestamp: 1000 },
|
||||
{ sender: "Friend", body: "current text", messageId: "msg-1", timestamp: 2000 },
|
||||
],
|
||||
});
|
||||
|
||||
const account = createMockAccount({ dmHistoryLimit: 2 });
|
||||
const config: OpenClawConfig = {};
|
||||
const core = createMockRuntime();
|
||||
setBlueBubblesRuntime(core);
|
||||
|
||||
unregister = registerBlueBubblesWebhookTarget({
|
||||
account,
|
||||
config,
|
||||
runtime: { log: vi.fn(), error: vi.fn() },
|
||||
core,
|
||||
path: "/bluebubbles-webhook",
|
||||
});
|
||||
|
||||
const req = createMockRequest("POST", "/bluebubbles-webhook", {
|
||||
type: "new-message",
|
||||
data: {
|
||||
text: "current text",
|
||||
handle: { address: "+15551234567" },
|
||||
isGroup: false,
|
||||
isFromMe: false,
|
||||
guid: "msg-1",
|
||||
chatGuid: "iMessage;-;+15550002002",
|
||||
date: Date.now(),
|
||||
},
|
||||
});
|
||||
const res = createMockResponse();
|
||||
|
||||
await handleBlueBubblesWebhookRequest(req, res);
|
||||
await flushAsync();
|
||||
|
||||
const callArgs = getFirstDispatchCall();
|
||||
const inboundHistory = (callArgs.ctx.InboundHistory ?? []) as Array<{ body: string }>;
|
||||
expect(inboundHistory).toHaveLength(2);
|
||||
expect(inboundHistory.map((entry) => entry.body)).toEqual(["older context", "current text"]);
|
||||
expect(inboundHistory.filter((entry) => entry.body === "current text")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("uses exponential backoff for unresolved backfill and stops after resolve", async () => {
|
||||
mockFetchBlueBubblesHistory
|
||||
.mockResolvedValueOnce({ resolved: false, entries: [] })
|
||||
.mockResolvedValueOnce({
|
||||
resolved: true,
|
||||
entries: [
|
||||
{ sender: "Friend", body: "older context", messageId: "hist-1", timestamp: 1000 },
|
||||
],
|
||||
});
|
||||
|
||||
const account = createMockAccount({ dmHistoryLimit: 4 });
|
||||
const config: OpenClawConfig = {};
|
||||
const core = createMockRuntime();
|
||||
setBlueBubblesRuntime(core);
|
||||
|
||||
unregister = registerBlueBubblesWebhookTarget({
|
||||
account,
|
||||
config,
|
||||
runtime: { log: vi.fn(), error: vi.fn() },
|
||||
core,
|
||||
path: "/bluebubbles-webhook",
|
||||
});
|
||||
|
||||
const mkPayload = (guid: string, text: string, now: number) => ({
|
||||
type: "new-message",
|
||||
data: {
|
||||
text,
|
||||
handle: { address: "+15551234567" },
|
||||
isGroup: false,
|
||||
isFromMe: false,
|
||||
guid,
|
||||
chatGuid: "iMessage;-;+15550003003",
|
||||
date: now,
|
||||
},
|
||||
});
|
||||
|
||||
let now = 1_700_000_000_000;
|
||||
const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now);
|
||||
try {
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook", mkPayload("msg-1", "first text", now)),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
expect(mockFetchBlueBubblesHistory).toHaveBeenCalledTimes(1);
|
||||
|
||||
now += 1_000;
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook", mkPayload("msg-2", "second text", now)),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
expect(mockFetchBlueBubblesHistory).toHaveBeenCalledTimes(1);
|
||||
|
||||
now += 6_000;
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook", mkPayload("msg-3", "third text", now)),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
expect(mockFetchBlueBubblesHistory).toHaveBeenCalledTimes(2);
|
||||
|
||||
const thirdCall = mockDispatchReplyWithBufferedBlockDispatcher.mock.calls[2]?.[0];
|
||||
const thirdHistory = (thirdCall?.ctx.InboundHistory ?? []) as Array<{ body: string }>;
|
||||
expect(thirdHistory.map((entry) => entry.body)).toContain("older context");
|
||||
expect(thirdHistory.map((entry) => entry.body)).toContain("third text");
|
||||
|
||||
now += 10_000;
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook", mkPayload("msg-4", "fourth text", now)),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
expect(mockFetchBlueBubblesHistory).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
nowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("caps inbound history payload size to reduce prompt-bomb risk", async () => {
|
||||
const huge = "x".repeat(8_000);
|
||||
mockFetchBlueBubblesHistory.mockResolvedValueOnce({
|
||||
resolved: true,
|
||||
entries: Array.from({ length: 20 }, (_, idx) => ({
|
||||
sender: `Friend ${idx}`,
|
||||
body: `${huge} ${idx}`,
|
||||
messageId: `hist-${idx}`,
|
||||
timestamp: idx + 1,
|
||||
})),
|
||||
});
|
||||
|
||||
const account = createMockAccount({ dmHistoryLimit: 20 });
|
||||
const config: OpenClawConfig = {};
|
||||
const core = createMockRuntime();
|
||||
setBlueBubblesRuntime(core);
|
||||
|
||||
unregister = registerBlueBubblesWebhookTarget({
|
||||
account,
|
||||
config,
|
||||
runtime: { log: vi.fn(), error: vi.fn() },
|
||||
core,
|
||||
path: "/bluebubbles-webhook",
|
||||
});
|
||||
|
||||
await handleBlueBubblesWebhookRequest(
|
||||
createMockRequest("POST", "/bluebubbles-webhook", {
|
||||
type: "new-message",
|
||||
data: {
|
||||
text: "latest text",
|
||||
handle: { address: "+15551234567" },
|
||||
isGroup: false,
|
||||
isFromMe: false,
|
||||
guid: "msg-bomb-1",
|
||||
chatGuid: "iMessage;-;+15550004004",
|
||||
date: Date.now(),
|
||||
},
|
||||
}),
|
||||
createMockResponse(),
|
||||
);
|
||||
await flushAsync();
|
||||
|
||||
const callArgs = getFirstDispatchCall();
|
||||
const inboundHistory = (callArgs.ctx.InboundHistory ?? []) as Array<{ body: string }>;
|
||||
const totalChars = inboundHistory.reduce((sum, entry) => sum + entry.body.length, 0);
|
||||
expect(inboundHistory.length).toBeLessThan(20);
|
||||
expect(totalChars).toBeLessThanOrEqual(12_000);
|
||||
expect(inboundHistory.every((entry) => entry.body.length <= 1_203)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("fromMe messages", () => {
|
||||
it("ignores messages from self (fromMe=true)", async () => {
|
||||
const account = createMockAccount();
|
||||
|
||||
@@ -211,6 +211,7 @@ export {
|
||||
clearHistoryEntries,
|
||||
clearHistoryEntriesIfEnabled,
|
||||
DEFAULT_GROUP_HISTORY_LIMIT,
|
||||
evictOldHistoryKeys,
|
||||
recordPendingHistoryEntry,
|
||||
recordPendingHistoryEntryIfEnabled,
|
||||
} from "../auto-reply/reply/history.js";
|
||||
|
||||
Reference in New Issue
Block a user