File size: 4,729 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 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | import { createHash } from "node:crypto";
import {
assertCanInsertPluginStateEntry,
bindPluginStateEntry,
createPluginStateError,
deleteExpiredPluginStateEntries,
deletePluginStateEntry,
enforcePostRegisterLimits,
parseStoredJson,
resolvePluginStateExpiresAtMs,
selectPluginStateEntry,
upsertPluginStateEntry,
type PluginStateDatabase,
type PluginStateRegisterEntryParams,
type PluginStateReadRow,
} from "./plugin-state-store.kernel.js";
import type {
PluginStateCompareResult,
PluginStateObservation,
PluginStateStoreOperation,
} from "./plugin-state-store.types.js";
type Key = { pluginId: string; namespace: string; key: string };
export type PluginStatePreparedComparison = Key & { comparison: string } & (
| { operation: "update"; action: "set"; valueJson: string; ttlMs?: number }
| { operation: "update" | "delete"; action: "keep" }
| { operation: "delete"; action: "delete" }
);
export type PluginStateComparisonLimits = Pick<
PluginStateRegisterEntryParams,
"maxEntries" | "overflowPolicy"
> & { maxPluginEntries: number };
const COMPARISON_PATTERN = /^1:([a-f0-9]{64}):([a-f0-9]{64}|-)$/u;
export function validatePluginStateComparison(
value: string,
operation: PluginStateStoreOperation,
): string {
const match = typeof value === "string" ? COMPARISON_PATTERN.exec(value) : null;
const scope = match?.[1];
if (!scope) {
throw createPluginStateError({
code: "PLUGIN_STATE_INVALID_INPUT",
operation,
message: "Plugin state comparison must be an observation returned by this store.",
});
}
return scope;
}
function digest(value: readonly unknown[]): string {
return createHash("sha256").update(JSON.stringify(value)).digest("hex");
}
function comparisonScope(storeIdentity: string, key: Key): string {
return digest([storeIdentity, key.pluginId, key.namespace, key.key]);
}
function observation(
store: PluginStateDatabase,
scope: string,
row: PluginStateReadRow | undefined,
operation: PluginStateStoreOperation,
): PluginStateObservation<unknown> {
// Preserve the stored JSON image; caller reserialization can change legacy whitespace/key order.
const image = row ? digest([row.value_json, row.created_at, row.expires_at]) : "-";
return {
value: row ? parseStoredJson(row.value_json, operation, store.path) : undefined,
comparison: `1:${scope}:${image}`,
};
}
/** Called after canonical writable admission, with the native owner's recorded database identity. */
export function observePluginStateEntry(
store: PluginStateDatabase,
params: Key,
storeIdentity: string,
): PluginStateObservation<unknown> {
return observation(
store,
comparisonScope(storeIdentity, params),
selectPluginStateEntry(store.db, { ...params, now: Date.now() }),
"lookup",
);
}
/** The caller owns the IMMEDIATE transaction containing comparison, expiry, quotas and mutation. */
export function compareAndApplyPluginStateEntry(
store: PluginStateDatabase,
params: PluginStatePreparedComparison & PluginStateComparisonLimits,
storeIdentity: string,
): PluginStateCompareResult<unknown> {
const operation = params.operation === "update" ? "register" : "delete";
const expected = validatePluginStateComparison(params.comparison, operation);
const scope = comparisonScope(storeIdentity, params);
if (expected !== scope) {
throw createPluginStateError({
code: "PLUGIN_STATE_INVALID_INPUT",
operation,
path: store.path,
message: "Plugin state observation belongs to another database, namespace or key.",
});
}
const now = Date.now();
const row = selectPluginStateEntry(store.db, { ...params, now });
const current = observation(
store,
scope,
row,
params.operation === "update" ? "lookup" : "delete",
);
if (current.comparison !== params.comparison) {
return { status: "conflict", current };
}
if (params.operation === "delete") {
return {
status:
params.action === "delete" && row && deletePluginStateEntry(store.db, params) > 0
? "applied"
: "unchanged",
};
}
deleteExpiredPluginStateEntries(store.db, now, params);
if (params.action === "keep") {
return { status: "unchanged" };
}
if (!row) {
assertCanInsertPluginStateEntry({ ...params, store, now });
}
const expiresAt = resolvePluginStateExpiresAtMs({
ttlMs: params.ttlMs,
now,
operation: "register",
path: store.path,
});
upsertPluginStateEntry(
store.db,
bindPluginStateEntry({
...params,
createdAt: now,
expiresAt,
}),
);
enforcePostRegisterLimits({ ...params, store, now, protectedKey: params.key });
return { status: "applied" };
}
|