mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-22 02:31:45 +00:00
fix(mattermost): preserve websocket posts across restarts (#110386)
* fix(mattermost): adopt durable ingress drain at the websocket chokepoint Posted events processed detached from the websocket receive with only a 5-minute in-memory guard; a crash lost the post and reconnect never replays it. Raw posted envelopes now journal durably (event_id = post.id per the upstream Post model, lane per channel_id, one row per post) before handler scheduling; dispatch runs through the core drain with deferred claims through debounce, merged-flush fan-out adoption, gated-turn settlement, and 30d/20k tombstones covering the old 5min/2k guard, which is deleted after its parity test. post_edited stays excluded and cannot be swallowed by posted tombstones. Cold-gap limitation stated: Mattermost cannot replay posts missed while disconnected. Autoreview blocked by codex sandbox network in the build stage; full manual review performed (updated one websocket test asserting the pre-adoption parsed-post contract to the raw-envelope contract). Part of #109657 wave 2. * style(mattermost): keep ingress monitor type internal * fix(mattermost): retry then loudly escalate a failed durable append Landing autoreview caught a real loss path: a durable enqueue failure at the websocket chokepoint was logged and swallowed — the raw envelope discarded, the connection kept running against a broken store, and reconnect never replays, so a transient SQLite failure silently lost the post. The append now retries with short backoff for transient blips; a persistent failure propagates and the websocket terminates loudly so the outage is operator-visible instead of silently dropping every subsequent post. Regression test covers both the absorbed-transient and escalation paths. * style(mattermost): format rebased ingress handler * docs(mattermost): document bounded auth-failure retries under deferred claims * fix(mattermost): guard the drain pump against stop racing the async prune stop() disposing before the startup pump finished pruning let the pump lazily create a fresh undisposed drain and dispatch after shutdown. The pump now re-checks running after the prune, and stop() disposes again after awaiting the pump so a drain created mid-race is torn down. Regression test blocks the prune across stop and asserts no dispatch. * fix(mattermost): serialize durable admissions to preserve lane order Concurrent websocket callbacks let a post in append-retry backoff be overtaken by its successor, inverting same-channel arrival order in the queue. Admissions now chain (order over latency, mirroring the iMessage admission tail); regression proves a retried post still lands ahead of a concurrently received one. * test(mattermost): assert lane order via dispatch sequence * fix(mattermost): stop() awaits in-flight admissions before disposal * fix(mattermost): satisfy ingress lint checks * fix(mattermost): honor envelope-level channel ids in the durable inspector Posts can carry their channel id on the post, the event data, or the broadcast envelope — the monitor dispatch honors all three, but the ingress inspector and claim-side validator required the nested field, rejecting valid posts as permanent and (via the storage-failure escalation) tearing down the socket for a failure that never happened, losing posts reconnect cannot replay. Both sites accept the three shapes; regression proves an envelope-level post dispatches.
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
746d257f05
commit
d4d23fe954
420
extensions/mattermost/src/mattermost/monitor-ingress.test.ts
Normal file
420
extensions/mattermost/src/mattermost/monitor-ingress.test.ts
Normal file
@@ -0,0 +1,420 @@
|
||||
// Mattermost durable ingress tests cover append, recovery, tombstones, and merged adoption.
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
createChannelIngressQueueForTests,
|
||||
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
buildMattermostFlushIngressLifecycle,
|
||||
createMattermostIngressMonitor,
|
||||
type MattermostIngressLifecycle,
|
||||
} from "./monitor-ingress.js";
|
||||
|
||||
type MattermostIngressQueue = NonNullable<
|
||||
Parameters<typeof createMattermostIngressMonitor>[0]["queue"]
|
||||
>;
|
||||
type MattermostIngressPayload = Parameters<MattermostIngressQueue["enqueue"]>[1];
|
||||
type MattermostIngressDispatch = Parameters<typeof createMattermostIngressMonitor>[0]["dispatch"];
|
||||
|
||||
function postedEvent(params?: {
|
||||
postId?: string;
|
||||
channelId?: string;
|
||||
message?: string;
|
||||
event?: "posted" | "post_edited";
|
||||
}): string {
|
||||
return JSON.stringify({
|
||||
event: params?.event ?? "posted",
|
||||
data: {
|
||||
post: JSON.stringify({
|
||||
id: params?.postId ?? "post-1",
|
||||
channel_id: params?.channelId ?? "channel-1",
|
||||
user_id: "user-1",
|
||||
message: params?.message ?? "hello",
|
||||
}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function startMonitor(queue: MattermostIngressQueue, dispatch: MattermostIngressDispatch) {
|
||||
return createMattermostIngressMonitor({
|
||||
accountId: "default",
|
||||
queue,
|
||||
dispatch,
|
||||
runtime: { error: vi.fn(), log: vi.fn() },
|
||||
pollIntervalMs: 60_000,
|
||||
adoptionStallTimeoutMs: 5_000,
|
||||
});
|
||||
}
|
||||
|
||||
async function withQueue<T>(fn: (queue: MattermostIngressQueue) => Promise<T>): Promise<T> {
|
||||
const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mattermost-ingress-"));
|
||||
const stateDir = await fs.realpath(created);
|
||||
const queue = createChannelIngressQueueForTests<MattermostIngressPayload>({
|
||||
channelId: "mattermost",
|
||||
accountId: "default",
|
||||
stateDir,
|
||||
});
|
||||
try {
|
||||
return await fn(queue);
|
||||
} finally {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await fs.rm(stateDir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
function testLifecycle() {
|
||||
const calls = {
|
||||
adopted: vi.fn(async () => {}),
|
||||
deferred: vi.fn(),
|
||||
finalizing: vi.fn(),
|
||||
abandoned: vi.fn(async () => {}),
|
||||
};
|
||||
const lifecycle: MattermostIngressLifecycle = {
|
||||
abortSignal: new AbortController().signal,
|
||||
onAdopted: calls.adopted,
|
||||
onDeferred: calls.deferred,
|
||||
onAdoptionFinalizing: calls.finalizing,
|
||||
onAbandoned: calls.abandoned,
|
||||
};
|
||||
return { calls, lifecycle };
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("Mattermost durable ingress", () => {
|
||||
it("propagates durable append failure before handler scheduling", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const appendError = new Error("sqlite unavailable");
|
||||
const failingQueue = {
|
||||
...queue,
|
||||
enqueue: vi.fn().mockRejectedValue(appendError),
|
||||
} satisfies MattermostIngressQueue;
|
||||
const dispatch = vi.fn();
|
||||
const monitor = startMonitor(failingQueue, dispatch);
|
||||
try {
|
||||
await expect(monitor.receive(postedEvent())).rejects.toBe(appendError);
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("recovers an uncompleted post with a fresh drain and dispatches exactly once", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const interruptedDispatch = vi.fn((_post, _payload, lifecycle) => {
|
||||
lifecycle.onDeferred();
|
||||
return { kind: "deferred" } as const;
|
||||
});
|
||||
const interrupted = startMonitor(queue, interruptedDispatch);
|
||||
await interrupted.receive(postedEvent({ postId: "post-restart" }));
|
||||
await interrupted.waitForIdle();
|
||||
expect(await queue.listClaims()).toHaveLength(1);
|
||||
await interrupted.stop();
|
||||
|
||||
const recoveredDispatch = vi.fn(async (_post, _payload, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const recovered = startMonitor(queue, recoveredDispatch);
|
||||
try {
|
||||
await recovered.waitForIdle();
|
||||
expect(recoveredDispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await recovered.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("retains completion so a duplicate post id cannot dispatch twice", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async (_post, _payload, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
const event = postedEvent({ postId: "post-completed" });
|
||||
await monitor.receive(event);
|
||||
await monitor.waitForIdle();
|
||||
await monitor.receive(event);
|
||||
await monitor.waitForIdle();
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves every id from the retired merged-message guard key space", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async (_post, _payload, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive(postedEvent({ postId: "post-batch-first", message: "first" }));
|
||||
await monitor.receive(postedEvent({ postId: "post-batch-second", message: "second" }));
|
||||
await monitor.waitForIdle();
|
||||
|
||||
await monitor.receive(postedEvent({ postId: "post-batch-first", message: "first" }));
|
||||
await monitor.waitForIdle();
|
||||
|
||||
expect(dispatch).toHaveBeenCalledTimes(2);
|
||||
expect(dispatch.mock.calls.map(([post]) => post.id)).toEqual([
|
||||
"post-batch-first",
|
||||
"post-batch-second",
|
||||
]);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("stores the exact raw envelope in a per-channel lane", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const rawEvent = postedEvent({ postId: "post-raw", channelId: "channel-raw" });
|
||||
const dispatch = vi.fn((_post, _payload, lifecycle) => {
|
||||
lifecycle.onDeferred();
|
||||
return { kind: "deferred" } as const;
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive(rawEvent);
|
||||
await monitor.waitForIdle();
|
||||
expect(await queue.listClaims()).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "post-raw",
|
||||
laneKey: "channel:channel-raw",
|
||||
payload: expect.objectContaining({ rawEvent }),
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("does not let ignored post_edited events tombstone a later posted event", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async (_post, _payload, lifecycle) => {
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive(postedEvent({ postId: "post-edit", event: "post_edited" }));
|
||||
await monitor.waitForIdle();
|
||||
await monitor.receive(postedEvent({ postId: "post-edit", event: "posted" }));
|
||||
await monitor.waitForIdle();
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves same-channel arrival order across an append retry backoff", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
let failFirst = true;
|
||||
const enqueue = queue.enqueue.bind(queue);
|
||||
queue.enqueue = async (...args) => {
|
||||
if (failFirst) {
|
||||
failFirst = false;
|
||||
throw new Error("sqlite busy");
|
||||
}
|
||||
return await enqueue(...args);
|
||||
};
|
||||
const dispatched: string[] = [];
|
||||
const ingress = startMonitor(queue, async (post, _payload, lifecycle) => {
|
||||
dispatched.push(post.id);
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
try {
|
||||
// A's first append fails into backoff; B arrives concurrently. The
|
||||
// admission chain must hold B until A commits, or lane order inverts.
|
||||
const admitA = ingress.receive(postedEvent({ postId: "post-A" }));
|
||||
const admitB = ingress.receive(postedEvent({ postId: "post-B" }));
|
||||
await Promise.all([admitA, admitB]);
|
||||
await vi.waitFor(() => expect(dispatched).toEqual(["post-A", "post-B"]), {
|
||||
timeout: 5_000,
|
||||
});
|
||||
} finally {
|
||||
await ingress.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("never dispatches from a pump racing stop through the async prune", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
let releasePrune = () => {};
|
||||
const pruneGate = new Promise<void>((resolve) => {
|
||||
releasePrune = resolve;
|
||||
});
|
||||
const prune = queue.prune.bind(queue);
|
||||
queue.prune = async (...args) => {
|
||||
await pruneGate;
|
||||
return await prune(...args);
|
||||
};
|
||||
const dispatch = vi.fn();
|
||||
const ingress = startMonitor(queue, dispatch);
|
||||
await ingress.receive(postedEvent({ postId: "post-stop-race" }));
|
||||
|
||||
const stopping = ingress.stop();
|
||||
releasePrune();
|
||||
await stopping;
|
||||
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
it("retries a transient append failure and escalates a persistent one", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
let failures = 2;
|
||||
const enqueue = queue.enqueue.bind(queue);
|
||||
queue.enqueue = async (...args) => {
|
||||
if (failures > 0) {
|
||||
failures -= 1;
|
||||
throw new Error("sqlite busy");
|
||||
}
|
||||
return await enqueue(...args);
|
||||
};
|
||||
const ingress = startMonitor(queue, vi.fn());
|
||||
try {
|
||||
// Two transient failures absorb into the bounded retry.
|
||||
await ingress.receive(postedEvent({ postId: "post-retry" }));
|
||||
expect(await queue.listPending({ limit: "all" })).toHaveLength(1);
|
||||
|
||||
// A persistent failure escalates so the websocket can tear down loudly.
|
||||
failures = Number.POSITIVE_INFINITY;
|
||||
await expect(ingress.receive(postedEvent({ postId: "post-lost" }))).rejects.toThrow(
|
||||
"sqlite busy",
|
||||
);
|
||||
} finally {
|
||||
await ingress.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("accepts a post whose channel id arrives at the envelope level", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatched: string[] = [];
|
||||
const ingress = startMonitor(queue, async (post, _payload, lifecycle) => {
|
||||
dispatched.push(post.id);
|
||||
await lifecycle.onAdopted();
|
||||
});
|
||||
try {
|
||||
// The monitor dispatch honors post/data/broadcast channel ids; the
|
||||
// durable inspector must not reject the envelope-level shapes.
|
||||
await ingress.receive(
|
||||
JSON.stringify({
|
||||
event: "posted",
|
||||
data: {
|
||||
channel_id: "channel-envelope",
|
||||
post: JSON.stringify({ id: "post-envelope", user_id: "user-1", message: "hi" }),
|
||||
},
|
||||
}),
|
||||
);
|
||||
await vi.waitFor(() => expect(dispatched).toEqual(["post-envelope"]));
|
||||
} finally {
|
||||
await ingress.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("dead-letters malformed persisted payloads without retry", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
await queue.enqueue(
|
||||
"post-malformed",
|
||||
{ version: 1, receivedAt: 1, rawEvent: "{" },
|
||||
{ receivedAt: 1, laneKey: "channel:channel-1" },
|
||||
);
|
||||
const dispatch = vi.fn();
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.waitForIdle();
|
||||
expect((await queue.enqueue("post-malformed", {} as MattermostIngressPayload)).kind).toBe(
|
||||
"failed",
|
||||
);
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("dead-letters permanent Mattermost auth failures", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async () => {
|
||||
throw new Error("Mattermost API 401 Unauthorized: invalid token");
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive(postedEvent({ postId: "post-auth" }));
|
||||
await monitor.waitForIdle();
|
||||
expect((await queue.enqueue("post-auth", {} as MattermostIngressPayload)).kind).toBe(
|
||||
"failed",
|
||||
);
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("leaves transient Mattermost failures retryable", async () => {
|
||||
await withQueue(async (queue) => {
|
||||
const dispatch = vi.fn(async () => {
|
||||
throw new Error("Mattermost API 503 Service Unavailable");
|
||||
});
|
||||
const monitor = startMonitor(queue, dispatch);
|
||||
try {
|
||||
await monitor.receive(postedEvent({ postId: "post-transient" }));
|
||||
await monitor.waitForIdle();
|
||||
expect(await queue.listPending({ limit: "all" })).toEqual([
|
||||
expect.objectContaining({ id: "post-transient" }),
|
||||
]);
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
await monitor.stop();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("Mattermost merged ingress lifecycle", () => {
|
||||
it("fans adoption out to every constituent claim", async () => {
|
||||
const first = testLifecycle();
|
||||
const second = testLifecycle();
|
||||
const merged = buildMattermostFlushIngressLifecycle([
|
||||
{ turnAdoptionLifecycle: first.lifecycle },
|
||||
{ turnAdoptionLifecycle: second.lifecycle },
|
||||
]);
|
||||
|
||||
merged.lifecycle?.onDeferred();
|
||||
await merged.lifecycle?.onAdopted();
|
||||
await merged.settle();
|
||||
|
||||
expect(first.calls.deferred).toHaveBeenCalledTimes(1);
|
||||
expect(second.calls.deferred).toHaveBeenCalledTimes(1);
|
||||
expect(first.calls.adopted).toHaveBeenCalledTimes(1);
|
||||
expect(second.calls.adopted).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("completes all claims when a gated flush never dispatches", async () => {
|
||||
const first = testLifecycle();
|
||||
const second = testLifecycle();
|
||||
const merged = buildMattermostFlushIngressLifecycle([
|
||||
{ turnAdoptionLifecycle: first.lifecycle },
|
||||
{ turnAdoptionLifecycle: second.lifecycle },
|
||||
]);
|
||||
|
||||
await merged.settle();
|
||||
|
||||
expect(first.calls.adopted).toHaveBeenCalledTimes(1);
|
||||
expect(second.calls.adopted).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
413
extensions/mattermost/src/mattermost/monitor-ingress.ts
Normal file
413
extensions/mattermost/src/mattermost/monitor-ingress.ts
Normal file
@@ -0,0 +1,413 @@
|
||||
// Mattermost plugin module owns raw WebSocket durable ingress mapping and draining.
|
||||
import {
|
||||
createChannelIngressDrain,
|
||||
DEFAULT_INGRESS_ADOPTION_STALL_MS,
|
||||
DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
|
||||
DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
|
||||
type ChannelIngressDrain,
|
||||
type ChannelIngressQueue,
|
||||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env";
|
||||
import { getMattermostRuntime } from "../runtime.js";
|
||||
import type { MattermostPost } from "./client.js";
|
||||
import {
|
||||
parseMattermostEventPayload,
|
||||
parseMattermostPost,
|
||||
type MattermostEventPayload,
|
||||
} from "./monitor-websocket.js";
|
||||
|
||||
const MATTERMOST_INGRESS_PAYLOAD_VERSION = 1;
|
||||
const MATTERMOST_INGRESS_POLL_INTERVAL_MS = 1_000;
|
||||
const MATTERMOST_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000;
|
||||
const MATTERMOST_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
|
||||
const MATTERMOST_INGRESS_COMPLETED_MAX_ENTRIES = 20_000;
|
||||
const MATTERMOST_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000;
|
||||
const MATTERMOST_INGRESS_FAILED_MAX_ENTRIES = 20_000;
|
||||
|
||||
export type MattermostIngressLifecycle = {
|
||||
abortSignal: AbortSignal;
|
||||
onAdopted: () => void | Promise<void>;
|
||||
onDeferred: () => void;
|
||||
onAdoptionFinalizing: () => void;
|
||||
onAbandoned: () => void | Promise<void>;
|
||||
};
|
||||
|
||||
/** Fan one merged Mattermost turn's adoption lifecycle across every source claim. */
|
||||
export function buildMattermostFlushIngressLifecycle(
|
||||
entries: ReadonlyArray<{ turnAdoptionLifecycle?: MattermostIngressLifecycle }>,
|
||||
): {
|
||||
lifecycle: MattermostIngressLifecycle | undefined;
|
||||
settle: () => Promise<void>;
|
||||
} {
|
||||
const lifecycles = entries
|
||||
.map((entry) => entry.turnAdoptionLifecycle)
|
||||
.filter((lifecycle) => lifecycle !== undefined);
|
||||
const [firstLifecycle] = lifecycles;
|
||||
if (!firstLifecycle) {
|
||||
return { lifecycle: undefined, settle: async () => {} };
|
||||
}
|
||||
let handedOff = false;
|
||||
const adoptAll = async () => {
|
||||
for (const lifecycle of lifecycles) {
|
||||
await lifecycle.onAdopted();
|
||||
}
|
||||
};
|
||||
return {
|
||||
lifecycle: {
|
||||
abortSignal:
|
||||
lifecycles.length === 1
|
||||
? firstLifecycle.abortSignal
|
||||
: AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)),
|
||||
onAdopted: async () => {
|
||||
handedOff = true;
|
||||
await adoptAll();
|
||||
},
|
||||
onDeferred: () => {
|
||||
handedOff = true;
|
||||
for (const lifecycle of lifecycles) {
|
||||
lifecycle.onDeferred();
|
||||
}
|
||||
},
|
||||
onAdoptionFinalizing: () => {
|
||||
for (const lifecycle of lifecycles) {
|
||||
lifecycle.onAdoptionFinalizing();
|
||||
}
|
||||
},
|
||||
onAbandoned: async () => {
|
||||
handedOff = true;
|
||||
await Promise.all(
|
||||
lifecycles.map(async (lifecycle) => {
|
||||
await lifecycle.onAbandoned();
|
||||
}),
|
||||
);
|
||||
},
|
||||
},
|
||||
// Gated/no-dispatch turns are terminal and must not leave source claims deferred.
|
||||
settle: async () => {
|
||||
if (!handedOff) {
|
||||
await adoptAll();
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
type MattermostIngressPayload = {
|
||||
version: 1;
|
||||
receivedAt: number;
|
||||
rawEvent: string;
|
||||
};
|
||||
|
||||
type MattermostIngressDispatchResult =
|
||||
| { kind: "completed" }
|
||||
| { kind: "deferred" }
|
||||
| { kind: "failed-retryable"; error: unknown };
|
||||
|
||||
type MattermostIngressDispatch = (
|
||||
post: MattermostPost,
|
||||
payload: MattermostEventPayload,
|
||||
lifecycle: MattermostIngressLifecycle,
|
||||
) => Promise<MattermostIngressDispatchResult | void> | MattermostIngressDispatchResult | void;
|
||||
|
||||
class MattermostIngressPermanentError extends Error {
|
||||
constructor(
|
||||
readonly reason: "invalid-event" | "mattermost-auth",
|
||||
message: string,
|
||||
options?: ErrorOptions,
|
||||
) {
|
||||
super(message, options);
|
||||
this.name = "MattermostIngressPermanentError";
|
||||
}
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function parseRawObject(raw: string, subject: string): Record<string, unknown> {
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(raw);
|
||||
} catch (error) {
|
||||
throw new MattermostIngressPermanentError(
|
||||
"invalid-event",
|
||||
`${subject} contains invalid JSON.`,
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
if (!isRecord(parsed)) {
|
||||
throw new MattermostIngressPermanentError("invalid-event", `${subject} must be a JSON object.`);
|
||||
}
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function parseRawPost(value: unknown): Record<string, unknown> {
|
||||
if (typeof value === "string") {
|
||||
return parseRawObject(value, "Mattermost posted event post");
|
||||
}
|
||||
if (isRecord(value)) {
|
||||
return value;
|
||||
}
|
||||
throw new MattermostIngressPermanentError(
|
||||
"invalid-event",
|
||||
"Mattermost posted event is missing its post object.",
|
||||
);
|
||||
}
|
||||
|
||||
function requiredString(value: unknown, field: string): string {
|
||||
if (typeof value === "string" && value.trim()) {
|
||||
return value.trim();
|
||||
}
|
||||
throw new MattermostIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Mattermost posted event is missing ${field}.`,
|
||||
);
|
||||
}
|
||||
|
||||
function inspectMattermostIngressEvent(rawEvent: string): {
|
||||
eventId: string;
|
||||
laneKey: string;
|
||||
} | null {
|
||||
const envelope = parseRawObject(rawEvent, "Mattermost WebSocket event");
|
||||
if (envelope.event !== "posted") {
|
||||
return null;
|
||||
}
|
||||
const data = isRecord(envelope.data) ? envelope.data : null;
|
||||
const post = parseRawPost(data?.post);
|
||||
const eventId = requiredString(post.id, "post.id");
|
||||
// Mattermost can carry the channel id on the post, the event data, or the
|
||||
// broadcast envelope (the monitor dispatch honors all three). Rejecting the
|
||||
// envelope-level shapes as permanent would drop valid posts and tear the
|
||||
// socket down for a storage failure that never happened.
|
||||
const broadcast = isRecord(envelope.broadcast) ? envelope.broadcast : null;
|
||||
const channelId =
|
||||
typeof post.channel_id === "string" && post.channel_id.trim()
|
||||
? post.channel_id.trim()
|
||||
: typeof data?.channel_id === "string" && data.channel_id.trim()
|
||||
? data.channel_id.trim()
|
||||
: requiredString(broadcast?.channel_id, "channel_id");
|
||||
return { eventId, laneKey: `channel:${channelId}` };
|
||||
}
|
||||
|
||||
function parseClaimedEvent(
|
||||
rawEvent: string,
|
||||
eventId: string,
|
||||
): {
|
||||
post: MattermostPost;
|
||||
payload: MattermostEventPayload;
|
||||
} {
|
||||
const payload = parseMattermostEventPayload(rawEvent);
|
||||
if (!payload || payload.event !== "posted") {
|
||||
throw new MattermostIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Mattermost ingress row ${eventId} is not a posted event.`,
|
||||
);
|
||||
}
|
||||
const post = parseMattermostPost(payload.data?.post);
|
||||
// Channel id may live on the post, the event data, or the broadcast — the
|
||||
// durable inspector accepted all three, so the claim-side check must too.
|
||||
const claimedChannelId =
|
||||
post?.channel_id?.trim() ||
|
||||
payload.data?.channel_id?.trim() ||
|
||||
payload.broadcast?.channel_id?.trim();
|
||||
if (!post || post.id !== eventId || !claimedChannelId) {
|
||||
throw new MattermostIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Mattermost ingress row ${eventId} has invalid post identity.`,
|
||||
);
|
||||
}
|
||||
return { post, payload };
|
||||
}
|
||||
|
||||
function resolveMattermostIngressNonRetryableFailure(error: unknown) {
|
||||
if (error instanceof MattermostIngressPermanentError) {
|
||||
return { reason: error.reason, message: error.message };
|
||||
}
|
||||
const message = formatErrorMessage(error);
|
||||
return /Mattermost API (?:401|403)\b/.test(message)
|
||||
? { reason: "mattermost-auth", message }
|
||||
: null;
|
||||
}
|
||||
|
||||
type MattermostIngressMonitor = {
|
||||
receive: (rawEvent: string) => Promise<void>;
|
||||
stop: () => Promise<void>;
|
||||
waitForIdle: () => Promise<void>;
|
||||
};
|
||||
|
||||
export function createMattermostIngressMonitor(options: {
|
||||
accountId: string;
|
||||
queue?: ChannelIngressQueue<MattermostIngressPayload>;
|
||||
dispatch: MattermostIngressDispatch;
|
||||
runtime: Pick<RuntimeEnv, "error" | "log">;
|
||||
pollIntervalMs?: number;
|
||||
adoptionStallTimeoutMs?: number;
|
||||
abortSignal?: AbortSignal;
|
||||
}): MattermostIngressMonitor {
|
||||
let queue = options.queue;
|
||||
let drain: ChannelIngressDrain | undefined;
|
||||
let running = true;
|
||||
let requested = false;
|
||||
let pumping: Promise<void> | undefined;
|
||||
let lastPrunedAt = 0;
|
||||
|
||||
const getQueue = (): ChannelIngressQueue<MattermostIngressPayload> => {
|
||||
queue ??= getMattermostRuntime().state.openChannelIngressQueue<MattermostIngressPayload>({
|
||||
accountId: options.accountId,
|
||||
});
|
||||
return queue;
|
||||
};
|
||||
|
||||
const getDrain = (): ChannelIngressDrain => {
|
||||
drain ??= createChannelIngressDrain<MattermostIngressPayload>({
|
||||
queue: getQueue(),
|
||||
adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS,
|
||||
retryPolicy: {
|
||||
maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS,
|
||||
deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS,
|
||||
},
|
||||
resolveNonRetryableFailure: resolveMattermostIngressNonRetryableFailure,
|
||||
...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
|
||||
onLog: (message) => options.runtime.log?.(`mattermost ${message}`),
|
||||
dispatchClaimedEvent: async (record, lifecycle) => {
|
||||
if (record.payload.version !== MATTERMOST_INGRESS_PAYLOAD_VERSION) {
|
||||
throw new MattermostIngressPermanentError(
|
||||
"invalid-event",
|
||||
`Mattermost ingress row ${record.id} has an unsupported version.`,
|
||||
);
|
||||
}
|
||||
const { post, payload } = parseClaimedEvent(record.payload.rawEvent, record.id);
|
||||
return await options.dispatch(post, payload, lifecycle);
|
||||
},
|
||||
});
|
||||
return drain;
|
||||
};
|
||||
|
||||
const pruneIfDue = async (): Promise<void> => {
|
||||
const now = Date.now();
|
||||
if (now - lastPrunedAt < MATTERMOST_INGRESS_PRUNE_INTERVAL_MS) {
|
||||
return;
|
||||
}
|
||||
await getQueue().prune({
|
||||
completedTtlMs: MATTERMOST_INGRESS_COMPLETED_TTL_MS,
|
||||
completedMaxEntries: MATTERMOST_INGRESS_COMPLETED_MAX_ENTRIES,
|
||||
failedTtlMs: MATTERMOST_INGRESS_FAILED_TTL_MS,
|
||||
failedMaxEntries: MATTERMOST_INGRESS_FAILED_MAX_ENTRIES,
|
||||
now,
|
||||
});
|
||||
lastPrunedAt = now;
|
||||
};
|
||||
|
||||
const runPump = async (): Promise<void> => {
|
||||
try {
|
||||
for (;;) {
|
||||
requested = false;
|
||||
await pruneIfDue();
|
||||
// stop() may have run during the async prune; creating the lazy drain
|
||||
// now would leave an undisposed instance dispatching after stop.
|
||||
if (!running) {
|
||||
break;
|
||||
}
|
||||
const activeDrain = getDrain();
|
||||
const { started } = await activeDrain.drainOnce();
|
||||
await activeDrain.waitForIdle();
|
||||
if (!running || (!requested && started === 0)) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
options.runtime.error?.(`mattermost ingress drain failed: ${formatErrorMessage(error)}`);
|
||||
} finally {
|
||||
pumping = undefined;
|
||||
if (running && requested) {
|
||||
requestDrain();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const requestDrain = (): void => {
|
||||
requested = true;
|
||||
if (!running || pumping) {
|
||||
return;
|
||||
}
|
||||
pumping = runPump();
|
||||
};
|
||||
|
||||
const timer = setInterval(
|
||||
requestDrain,
|
||||
options.pollIntervalMs ?? MATTERMOST_INGRESS_POLL_INTERVAL_MS,
|
||||
);
|
||||
timer.unref?.();
|
||||
requestDrain();
|
||||
|
||||
// Serialize admissions: WS message callbacks run concurrently, and a post
|
||||
// stuck in append-retry backoff must delay its successors or same-channel
|
||||
// arrival order inverts in the queue (order over latency).
|
||||
let admissionTail: Promise<void> = Promise.resolve();
|
||||
|
||||
const admitOnce = async (rawEvent: string): Promise<void> => {
|
||||
const facts = inspectMattermostIngressEvent(rawEvent);
|
||||
if (!facts) {
|
||||
return;
|
||||
}
|
||||
const receivedAt = Date.now();
|
||||
// Websockets have no nack: a dropped append is a lost post (reconnect
|
||||
// never replays). Retry transient failures before letting the error
|
||||
// escalate to the socket teardown in monitor-websocket.
|
||||
let lastError: unknown;
|
||||
for (const delayMs of [0, 100, 300]) {
|
||||
if (delayMs > 0) {
|
||||
await new Promise((resolve) => {
|
||||
setTimeout(resolve, delayMs);
|
||||
});
|
||||
}
|
||||
try {
|
||||
await getQueue().enqueue(
|
||||
facts.eventId,
|
||||
{
|
||||
version: MATTERMOST_INGRESS_PAYLOAD_VERSION,
|
||||
receivedAt,
|
||||
rawEvent,
|
||||
},
|
||||
{ receivedAt, laneKey: facts.laneKey },
|
||||
);
|
||||
requestDrain();
|
||||
return;
|
||||
} catch (error) {
|
||||
lastError = error;
|
||||
}
|
||||
}
|
||||
throw lastError;
|
||||
};
|
||||
|
||||
return {
|
||||
receive: (rawEvent) => {
|
||||
const admission = admissionTail.then(() => admitOnce(rawEvent));
|
||||
admissionTail = admission.catch(() => undefined);
|
||||
return admission;
|
||||
},
|
||||
stop: async () => {
|
||||
running = false;
|
||||
clearInterval(timer);
|
||||
// A caller returning from stop() must know every accepted envelope is
|
||||
// durably committed; an in-flight admission racing process exit would
|
||||
// otherwise lose the post.
|
||||
await admissionTail;
|
||||
drain?.dispose();
|
||||
await pumping;
|
||||
// The pump may have lazily created the drain after the first dispose.
|
||||
drain?.dispose();
|
||||
await drain?.waitForIdle();
|
||||
},
|
||||
waitForIdle: async () => {
|
||||
for (;;) {
|
||||
const activePump = pumping;
|
||||
if (!activePump) {
|
||||
break;
|
||||
}
|
||||
await activePump;
|
||||
}
|
||||
await drain?.waitForIdle();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
// Mattermost plugin module owns replay-guarded post processing.
|
||||
import { createChannelReplayGuard } from "openclaw/plugin-sdk/persistent-dedupe";
|
||||
|
||||
const RECENT_MATTERMOST_MESSAGE_TTL_MS = 5 * 60_000;
|
||||
const RECENT_MATTERMOST_MESSAGE_MAX = 2000;
|
||||
function buildMattermostInboundReplayKeys(params: {
|
||||
accountId: string;
|
||||
messageIds: string[];
|
||||
}): string[] {
|
||||
return params.messageIds.map((id) => (id.trim() ? `${params.accountId}:${id.trim()}` : ""));
|
||||
}
|
||||
|
||||
function createMattermostInboundReplayGuard() {
|
||||
return createChannelReplayGuard<{ accountId: string; messageIds: string[] }>({
|
||||
dedupe: {
|
||||
ttlMs: RECENT_MATTERMOST_MESSAGE_TTL_MS,
|
||||
memoryMaxSize: RECENT_MATTERMOST_MESSAGE_MAX,
|
||||
},
|
||||
buildReplayKey: buildMattermostInboundReplayKeys,
|
||||
});
|
||||
}
|
||||
|
||||
type MattermostInboundReplayGuard = ReturnType<typeof createMattermostInboundReplayGuard>;
|
||||
const recentInboundMessages = createMattermostInboundReplayGuard();
|
||||
|
||||
export async function processMattermostReplayGuardedPost(params: {
|
||||
accountId: string;
|
||||
messageIds: string[];
|
||||
handlePost: () => Promise<void>;
|
||||
replayGuard?: MattermostInboundReplayGuard;
|
||||
}): Promise<"processed" | "duplicate"> {
|
||||
const replayGuard = params.replayGuard ?? recentInboundMessages;
|
||||
const event = {
|
||||
accountId: params.accountId,
|
||||
messageIds: params.messageIds,
|
||||
};
|
||||
const result = await replayGuard.processGuarded(event, params.handlePost, {
|
||||
onError: "commit",
|
||||
});
|
||||
return result.kind === "processed" ? "processed" : "duplicate";
|
||||
}
|
||||
@@ -284,14 +284,10 @@ describe("mattermost websocket monitor", () => {
|
||||
await once(server, "close");
|
||||
}
|
||||
|
||||
expect(onPosted).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ id: "post-1", message: "normal Mattermost post" }),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(onPosted).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ id: "post-large", props: largeProps }),
|
||||
expect.any(Object),
|
||||
);
|
||||
// onPosted now receives the raw envelope; the post rides inside it as a
|
||||
// nested JSON string, so its fields appear escaped.
|
||||
expect(onPosted).toHaveBeenCalledWith(expect.stringContaining('\\"id\\":\\"post-1\\"'));
|
||||
expect(onPosted).toHaveBeenCalledWith(expect.stringContaining('\\"id\\":\\"post-large\\"'));
|
||||
expect(runtime.error).toHaveBeenCalledWith(
|
||||
expect.stringContaining("Max payload size exceeded"),
|
||||
);
|
||||
@@ -347,6 +343,52 @@ describe("mattermost websocket monitor", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("hands posted envelopes to ingress raw and keeps post_edited out", async () => {
|
||||
const socket = new FakeWebSocket();
|
||||
const onPosted = vi.fn(async () => {});
|
||||
const connectOnce = createMattermostConnectOnce({
|
||||
wsUrl: "wss://example.invalid/api/v4/websocket",
|
||||
botToken: "token",
|
||||
runtime: testRuntime(),
|
||||
nextSeq: () => 1,
|
||||
onPosted,
|
||||
webSocketFactory: () => socket,
|
||||
});
|
||||
const posted = {
|
||||
event: "posted",
|
||||
data: {
|
||||
post: JSON.stringify({
|
||||
id: "post-raw",
|
||||
channel_id: "channel-raw",
|
||||
unexpected_transport_field: true,
|
||||
}),
|
||||
},
|
||||
};
|
||||
|
||||
const connected = connectOnce();
|
||||
socket.emitOpen();
|
||||
socket.emitMessage(
|
||||
Buffer.from(
|
||||
JSON.stringify({
|
||||
...posted,
|
||||
event: "post_edited",
|
||||
}),
|
||||
),
|
||||
);
|
||||
await new Promise<void>((resolve) => {
|
||||
queueMicrotask(resolve);
|
||||
});
|
||||
expect(onPosted).not.toHaveBeenCalled();
|
||||
socket.emitMessage(Buffer.from(JSON.stringify(posted)));
|
||||
await vi.waitFor(() => {
|
||||
expect(onPosted).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
socket.emitClose(1000);
|
||||
await connected;
|
||||
|
||||
expect(onPosted).toHaveBeenCalledWith(JSON.stringify(posted));
|
||||
});
|
||||
|
||||
it("terminates when bot update_at changes (disable/enable cycle)", async () => {
|
||||
vi.useFakeTimers();
|
||||
const socket = new FakeWebSocket();
|
||||
|
||||
@@ -15,7 +15,7 @@ import type { ChannelAccountSnapshot, RuntimeEnv } from "./runtime-api.js";
|
||||
export type MattermostEventPayload = {
|
||||
event?: string;
|
||||
data?: {
|
||||
post?: string | MattermostPost;
|
||||
post?: unknown;
|
||||
reaction?: string | Record<string, unknown>;
|
||||
channel_id?: string;
|
||||
channel_name?: string;
|
||||
@@ -58,7 +58,8 @@ const MattermostEventPayloadSchema = z.object({
|
||||
event: z.string().optional(),
|
||||
data: z
|
||||
.object({
|
||||
post: z.union([z.string(), MattermostPostSchema]).optional(),
|
||||
// Durable ingress validates the post only after claiming the raw envelope.
|
||||
post: z.unknown().optional(),
|
||||
reaction: z.union([z.string(), z.record(z.string(), z.unknown())]).optional(),
|
||||
channel_id: z.string().optional(),
|
||||
channel_name: z.string().optional(),
|
||||
@@ -77,11 +78,11 @@ const MattermostEventPayloadSchema = z.object({
|
||||
.optional(),
|
||||
}) as z.ZodType<MattermostEventPayload>;
|
||||
|
||||
function parseMattermostEventPayload(raw: string): MattermostEventPayload | null {
|
||||
export function parseMattermostEventPayload(raw: string): MattermostEventPayload | null {
|
||||
return safeParseJsonWithSchema(MattermostEventPayloadSchema, raw);
|
||||
}
|
||||
|
||||
function parseMattermostPost(value: unknown): MattermostPost | null {
|
||||
export function parseMattermostPost(value: unknown): MattermostPost | null {
|
||||
if (typeof value === "string") {
|
||||
return safeParseJsonWithSchema(MattermostPostSchema, value);
|
||||
}
|
||||
@@ -105,7 +106,7 @@ type CreateMattermostConnectOnceOpts = {
|
||||
statusSink?: (patch: Partial<ChannelAccountSnapshot>) => void;
|
||||
runtime: RuntimeEnv;
|
||||
nextSeq: () => number;
|
||||
onPosted: (post: MattermostPost, payload: MattermostEventPayload) => Promise<void>;
|
||||
onPosted: (rawEvent: string) => Promise<void>;
|
||||
onReaction?: (payload: MattermostEventPayload) => Promise<void>;
|
||||
webSocketFactory?: MattermostWebSocketFactory;
|
||||
/**
|
||||
@@ -129,23 +130,6 @@ const defaultMattermostWebSocketFactory: MattermostWebSocketFactory = (url, opti
|
||||
}) as MattermostWebSocketLike;
|
||||
};
|
||||
|
||||
function parsePostedPayload(
|
||||
payload: MattermostEventPayload,
|
||||
): { payload: MattermostEventPayload; post: MattermostPost } | null {
|
||||
if (payload.event !== "posted") {
|
||||
return null;
|
||||
}
|
||||
const postData = payload.data?.post;
|
||||
if (!postData) {
|
||||
return null;
|
||||
}
|
||||
const post = parseMattermostPost(postData);
|
||||
if (!post) {
|
||||
return null;
|
||||
}
|
||||
return { payload, post };
|
||||
}
|
||||
|
||||
export function createMattermostConnectOnce(
|
||||
opts: CreateMattermostConnectOnceOpts,
|
||||
): () => Promise<void> {
|
||||
@@ -375,14 +359,17 @@ export function createMattermostConnectOnce(
|
||||
if (payload.event !== "posted") {
|
||||
return;
|
||||
}
|
||||
const parsed = parsePostedPayload(payload);
|
||||
if (!parsed) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await opts.onPosted(parsed.post, parsed.payload);
|
||||
await opts.onPosted(raw);
|
||||
} catch (err) {
|
||||
opts.runtime.error?.(`mattermost handler failed: ${String(err)}`);
|
||||
// Durable admission failed after retries: this post is lost and the
|
||||
// websocket cannot nack or replay. Tear the connection down loudly
|
||||
// so the outage is operator-visible instead of silently dropping
|
||||
// every subsequent post against a broken store.
|
||||
opts.runtime.error?.(
|
||||
`mattermost durable admission failed; terminating websocket: ${String(err)}`,
|
||||
);
|
||||
ws.terminate();
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
// Mattermost tests cover monitor.inbound system event plugin behavior.
|
||||
import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { MattermostPost } from "./client.js";
|
||||
import type { MattermostEventPayload } from "./monitor-websocket.js";
|
||||
import { monitorMattermostProvider } from "./monitor.js";
|
||||
import type { OpenClawConfig, ReplyPayload, RuntimeEnv } from "./runtime-api.js";
|
||||
|
||||
@@ -144,6 +146,36 @@ vi.mock("./monitor-resources.js", async (importOriginal) => ({
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock("./monitor-ingress.js", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("./monitor-ingress.js")>();
|
||||
return {
|
||||
...actual,
|
||||
createMattermostIngressMonitor: (
|
||||
options: Parameters<typeof actual.createMattermostIngressMonitor>[0],
|
||||
) => ({
|
||||
receive: async (rawEvent: string) => {
|
||||
const payload = JSON.parse(rawEvent) as MattermostEventPayload;
|
||||
const post =
|
||||
typeof payload.data?.post === "string"
|
||||
? (JSON.parse(payload.data.post) as MattermostPost)
|
||||
: (payload.data?.post as MattermostPost | undefined);
|
||||
if (payload.event !== "posted" || !post) {
|
||||
return;
|
||||
}
|
||||
await options.dispatch(post, payload, {
|
||||
abortSignal: new AbortController().signal,
|
||||
onAdopted: async () => {},
|
||||
onDeferred: () => {},
|
||||
onAdoptionFinalizing: () => {},
|
||||
onAbandoned: async () => {},
|
||||
});
|
||||
},
|
||||
stop: async () => {},
|
||||
waitForIdle: async () => {},
|
||||
}),
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("./monitor-slash.js", () => ({
|
||||
registerMattermostMonitorSlashCommands: mockState.registerMattermostMonitorSlashCommands,
|
||||
}));
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user