Skip to content

Commit d2df089

Browse files
steipetesteipete
committed
refactor(matrix): persist ephemeral state in plugin sqlite
1 parent 522da25 commit d2df089

10 files changed

Lines changed: 774 additions & 298 deletions

‎extensions/matrix/src/matrix/client/storage.test.ts‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,11 @@ import { installMatrixTestRuntime } from "../../test-runtime.js";
77
import {
88
claimCurrentTokenStorageState,
99
maybeMigrateLegacyStorage,
10+
recordCurrentStorageMetaDeviceId,
1011
repairCurrentTokenStorageMetaDeviceId,
1112
resolveMatrixStateFilePath,
1213
resolveMatrixStoragePaths,
14+
writeStorageMeta,
1315
} from "./storage.js";
1416

1517
const createBackupArchiveMock = vi.hoisted(() =>
@@ -258,6 +260,35 @@ describe("matrix client storage paths", () => {
258260
fs.writeFileSync(path.join(rootDir, filename), JSON.stringify(value, null, 2));
259261
}
260262

263+
it("records a learned deviceId in storage metadata without startup JSON", () => {
264+
const stateDir = setupStateDir();
265+
const storagePaths = resolveMatrixStoragePaths({
266+
...defaultStorageAuth,
267+
env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
268+
});
269+
expect(
270+
writeStorageMeta({
271+
storagePaths,
272+
homeserver: defaultStorageAuth.homeserver,
273+
userId: defaultStorageAuth.userId,
274+
deviceId: null,
275+
}),
276+
).toBe(true);
277+
278+
expect(
279+
recordCurrentStorageMetaDeviceId({
280+
rootDir: storagePaths.rootDir,
281+
deviceId: "DEVICE123",
282+
}),
283+
).toBe(true);
284+
285+
const meta = JSON.parse(fs.readFileSync(storagePaths.metaPath, "utf8")) as {
286+
deviceId?: string | null;
287+
};
288+
expect(meta.deviceId).toBe("DEVICE123");
289+
expect(fs.existsSync(path.join(storagePaths.rootDir, "startup-verification.json"))).toBe(false);
290+
});
291+
261292
function seedExistingStorageRoot(params: {
262293
accessToken: string;
263294
deviceId?: string;

‎extensions/matrix/src/matrix/client/storage.ts‎

Lines changed: 23 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ const THREAD_BINDINGS_FILENAME = "thread-bindings.json";
2222
const LEGACY_CRYPTO_MIGRATION_FILENAME = "legacy-crypto-migration.json";
2323
const RECOVERY_KEY_FILENAME = "recovery-key.json";
2424
const IDB_SNAPSHOT_FILENAME = "crypto-idb-snapshot.json";
25-
const STARTUP_VERIFICATION_FILENAME = "startup-verification.json";
2625

2726
type LegacyMoveRecord = {
2827
sourcePath: string;
@@ -133,17 +132,6 @@ function readStoredRootMetadata(rootDir: string): StoredRootMetadata {
133132
}
134133
}
135134

136-
const verification = loadJsonFile<{ deviceId?: unknown }>(
137-
path.join(rootDir, STARTUP_VERIFICATION_FILENAME),
138-
);
139-
if (
140-
!metadata.deviceId &&
141-
typeof verification?.deviceId === "string" &&
142-
verification.deviceId.trim()
143-
) {
144-
metadata.deviceId = verification.deviceId.trim();
145-
}
146-
147135
return metadata;
148136
}
149137

@@ -516,6 +504,29 @@ export function claimCurrentTokenStorageState(params: { rootDir: string }): bool
516504
});
517505
}
518506

507+
export function recordCurrentStorageMetaDeviceId(params: {
508+
rootDir: string;
509+
deviceId: string;
510+
}): boolean {
511+
const deviceId = params.deviceId.trim();
512+
if (!deviceId) {
513+
return false;
514+
}
515+
const metadata = readStoredRootMetadata(params.rootDir);
516+
if (!metadata.accessTokenHash?.trim()) {
517+
return false;
518+
}
519+
return writeStoredRootMetadata(path.join(params.rootDir, STORAGE_META_FILENAME), {
520+
homeserver: metadata.homeserver,
521+
userId: metadata.userId,
522+
accountId: metadata.accountId ?? DEFAULT_ACCOUNT_KEY,
523+
accessTokenHash: metadata.accessTokenHash,
524+
deviceId,
525+
currentTokenStateClaimed: metadata.currentTokenStateClaimed === true,
526+
createdAt: metadata.createdAt ?? new Date().toISOString(),
527+
});
528+
}
529+
519530
export function repairCurrentTokenStorageMetaDeviceId(params: {
520531
homeserver: string;
521532
userId: string;

‎extensions/matrix/src/matrix/monitor/inbound-dedupe.test.ts‎

Lines changed: 82 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,33 @@
11
import fs from "node:fs";
22
import os from "node:os";
33
import path from "node:path";
4-
import { afterEach, describe, expect, it, vi } from "vitest";
4+
import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
5+
import {
6+
createPluginStateKeyedStoreForTests,
7+
resetPluginStateStoreForTests,
8+
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
9+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
10+
import type { PluginRuntime } from "../../runtime-api.js";
11+
import { setMatrixRuntime } from "../../runtime.js";
12+
import { LogService } from "../sdk/logger.js";
513
import { createMatrixInboundEventDeduper } from "./inbound-dedupe.js";
614

715
describe("Matrix inbound event dedupe", () => {
816
const tempDirs: string[] = [];
917

18+
beforeEach(() => {
19+
setMatrixRuntime({
20+
state: {
21+
openKeyedStore: (options: OpenKeyedStoreOptions) =>
22+
createPluginStateKeyedStoreForTests("matrix", options),
23+
},
24+
} as unknown as PluginRuntime);
25+
});
26+
1027
afterEach(() => {
1128
vi.restoreAllMocks();
1229
vi.useRealTimers();
30+
resetPluginStateStoreForTests();
1331
for (const dir of tempDirs.splice(0)) {
1432
fs.rmSync(dir, { recursive: true, force: true });
1533
}
@@ -125,22 +143,73 @@ describe("Matrix inbound event dedupe", () => {
125143
expect(second.claimEvent({ roomId: "!room:example.org", eventId: "$backlog" })).toBe(false);
126144
});
127145

128-
it("treats stop persistence failures as best-effort cleanup", async () => {
129-
const blockingPath = createStoragePath();
130-
fs.writeFileSync(blockingPath, "blocking file", "utf8");
131-
const deduper = await createMatrixInboundEventDeduper({
146+
it("imports legacy JSON entries into plugin state", async () => {
147+
const storagePath = createStoragePath();
148+
fs.writeFileSync(
149+
storagePath,
150+
JSON.stringify({
151+
version: 1,
152+
entries: [{ key: "!room:example.org|$legacy", ts: 90 }],
153+
}),
154+
"utf8",
155+
);
156+
157+
const first = await createMatrixInboundEventDeduper({
132158
auth: auth as never,
133-
storagePath: path.join(blockingPath, "nested", "inbound-dedupe.json"),
159+
storagePath,
160+
ttlMs: 20,
161+
nowMs: () => 100,
134162
});
163+
expect(first.claimEvent({ roomId: "!room:example.org", eventId: "$legacy" })).toBe(false);
135164

136-
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$persist-fail" })).toBe(
137-
true,
138-
);
139-
await deduper.commitEvent({
140-
roomId: "!room:example.org",
141-
eventId: "$persist-fail",
165+
fs.rmSync(storagePath, { force: true });
166+
resetPluginStateStoreForTests();
167+
const second = await createMatrixInboundEventDeduper({
168+
auth: auth as never,
169+
storagePath,
170+
ttlMs: 20,
171+
nowMs: () => 100,
172+
});
173+
expect(second.claimEvent({ roomId: "!room:example.org", eventId: "$legacy" })).toBe(false);
174+
});
175+
176+
it("keeps committed events in memory when plugin-state persistence fails", async () => {
177+
const storagePath = createStoragePath();
178+
const warnSpy = vi.spyOn(LogService, "warn").mockImplementation(() => {});
179+
setMatrixRuntime({
180+
state: {
181+
openKeyedStore: () => ({
182+
entries: async () => [],
183+
register: async () => {
184+
throw new Error("sqlite unavailable");
185+
},
186+
registerIfAbsent: async () => false,
187+
lookup: async () => undefined,
188+
consume: async () => undefined,
189+
delete: async () => false,
190+
clear: async () => {},
191+
}),
192+
},
193+
} as unknown as PluginRuntime);
194+
const deduper = await createMatrixInboundEventDeduper({
195+
auth: auth as never,
196+
storagePath,
142197
});
143198

144-
await expect(deduper.stop()).resolves.toBeUndefined();
199+
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$best-effort" })).toBe(true);
200+
await expect(
201+
deduper.commitEvent({
202+
roomId: "!room:example.org",
203+
eventId: "$best-effort",
204+
}),
205+
).resolves.toBeUndefined();
206+
expect(deduper.claimEvent({ roomId: "!room:example.org", eventId: "$best-effort" })).toBe(
207+
false,
208+
);
209+
expect(warnSpy).toHaveBeenCalledWith(
210+
"MatrixInboundDedupe",
211+
"Failed persisting Matrix inbound dedupe entry:",
212+
expect.any(Error),
213+
);
145214
});
146215
});

0 commit comments

Comments
 (0)