File size: 3,736 Bytes
76289e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | import { toUSVString } from "node:util";
import { err, ok, type Result } from "@openclaw/normalization-core/result";
import { executeSqliteQuerySync, sqliteStringSet } from "../infra/kysely-sync.js";
import {
createPluginStateError,
getPluginStateKysely,
selectPluginStateEntriesInKeyRange,
iteratePluginStateEntries,
parseStoredJson,
rowToEntry,
type PluginStateDatabase,
} from "./plugin-state-store.kernel.js";
import { PluginStateStoreError, type PluginStateEntry } from "./plugin-state-store.types.js";
export function lookupPluginStateEntries(
store: PluginStateDatabase,
params: { pluginId: string; namespace: string; keys: readonly string[] },
): Array<Result<unknown, PluginStateStoreError>> {
const now = Date.now();
const rows = executeSqliteQuerySync(
store.db,
getPluginStateKysely(store.db)
.selectFrom("plugin_state_entries")
.select(["entry_key", "value_json"])
.where("plugin_id", "=", params.pluginId)
.where("namespace", "=", params.namespace)
.where("entry_key", "in", sqliteStringSet(params.keys))
.where((eb) => eb.or([eb("expires_at", "is", null), eb("expires_at", ">", now)])),
).rows;
const values = new Map(rows.map((row) => [row.entry_key, row.value_json]));
return params.keys.map((key): Result<unknown, PluginStateStoreError> => {
// Match node:sqlite text binding, including lone UTF-16 surrogates.
const raw = values.get(toUSVString(key));
try {
return ok(raw === undefined ? undefined : parseStoredJson(raw, "lookup", store.path));
} catch (error) {
// Let ordered readers stop before a later corrupt value, just as with lookup.
if (error instanceof PluginStateStoreError && error.code === "PLUGIN_STATE_CORRUPT") {
return err(error);
}
throw error;
}
});
}
export function listPluginStateEntries(
store: PluginStateDatabase,
params: { pluginId: string; namespace: string },
): PluginStateEntry<unknown>[] {
const rows = iteratePluginStateEntries(store.db, {
pluginId: params.pluginId,
namespace: params.namespace,
now: Date.now(),
});
const entries: PluginStateEntry<unknown>[] = [];
let decodeFailure: { error: unknown } | undefined;
for (const row of rows) {
if (decodeFailure) {
continue;
}
try {
entries.push(rowToEntry(row, "entries", store.path));
} catch (error) {
// Finish the SQL read so a later step failure still precedes JSON errors.
decodeFailure = { error };
}
}
if (decodeFailure) {
throw decodeFailure.error;
}
return entries;
}
export type PluginStateKeyRangeParams = {
pluginId: string;
namespace: string;
keyStartInclusive: string;
keyEndExclusive: string;
limit: number;
order?: "asc" | "desc";
};
export function validatePluginStateKeyRange(params: PluginStateKeyRangeParams): void {
if (!Number.isSafeInteger(params.limit) || params.limit < 1) {
throw createPluginStateError({
code: "PLUGIN_STATE_INVALID_INPUT",
operation: "entries",
message: "Plugin state key-range limit must be a positive safe integer.",
});
}
if (params.keyStartInclusive >= params.keyEndExclusive) {
throw createPluginStateError({
code: "PLUGIN_STATE_INVALID_INPUT",
operation: "entries",
message: "Plugin state key range must have an increasing exclusive upper bound.",
});
}
}
export function listPluginStateEntriesInKeyRange(
store: PluginStateDatabase,
params: PluginStateKeyRangeParams,
): PluginStateEntry<unknown>[] {
return selectPluginStateEntriesInKeyRange(store.db, {
...params,
order: params.order ?? "asc",
now: Date.now(),
}).map((row) => rowToEntry(row, "entries", store.path));
}
|