Skip to content

Add typed plugin storage collections #870

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RhysSullivan merged 1 commit into from
May 29, 2026
Merged

Add typed plugin storage collections #870

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
258 changes: 258 additions & 0 deletions packages/core/sdk/src/executor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,8 +120,13 @@ import type {
} from "./plugin";
import {
pluginStorageId,
type PluginStorageCollectionData,
type PluginStorageCollectionDefinition,
type PluginStorageCollectionQueryInput,
type PluginStorageEntry,
type PluginStorageFacade,
type PluginStorageRuntimeCollectionDefinition,
type PluginStorageRuntimeIndexSpec,
} from "./plugin-storage";
import type { Scope } from "./scope";
import { RemoveSecretInput, SecretRef, SetSecretInput, type SecretProvider } from "./secrets";
Expand Down Expand Up @@ -801,6 +806,108 @@ const pluginStorageEntryFromRow = <T>(row: CoreRow<"plugin_storage">): PluginSto
updatedAt: row.updated_at instanceof Date ? row.updated_at : new Date(row.updated_at),
});

const pluginStorageIndexSpecFields = (spec: PluginStorageRuntimeIndexSpec): readonly string[] =>
typeof spec === "string" ? [spec] : spec;

const pluginStorageCollectionIndexedFields = (
definition: PluginStorageRuntimeCollectionDefinition,
): ReadonlySet<string> =>
new Set(definition.indexes.flatMap((spec) => pluginStorageIndexSpecFields(spec)));

const pluginStorageQueryValidationError = (
definition: PluginStorageRuntimeCollectionDefinition,
query: PluginStorageCollectionQueryInput<PluginStorageCollectionDefinition> | undefined,
): StorageError | null => {
if (!query) return null;

const indexedFields = pluginStorageCollectionIndexedFields(definition);
const fields = new Set<string>([
...Object.keys(query.where ?? {}),
...(query.orderBy ?? []).map((order) => order.field),
]);
for (const field of fields) {
if (!indexedFields.has(field)) {
return new StorageError({
message: `Plugin storage collection "${definition.name}" cannot query field "${field}" because it is not declared as an index`,
cause: undefined,
});
}
}

if (query.limit !== undefined && (!Number.isInteger(query.limit) || query.limit < 0)) {
return new StorageError({
message: `Plugin storage collection "${definition.name}" received an invalid query limit`,
cause: undefined,
});
}
if (query.offset !== undefined && (!Number.isInteger(query.offset) || query.offset < 0)) {
return new StorageError({
message: `Plugin storage collection "${definition.name}" received an invalid query offset`,
cause: undefined,
});
}

return null;
};

const isPluginStorageRecord = (value: unknown): value is Readonly<Record<string, unknown>> =>
value !== null && typeof value === "object" && !Array.isArray(value);

const pluginStorageWhereOperators = ["eq", "in", "gt", "gte", "lt", "lte"] as const;

const isPluginStorageWhereFilter = (value: unknown): value is Readonly<Record<string, unknown>> =>
isPluginStorageRecord(value) && pluginStorageWhereOperators.some((operator) => operator in value);

const pluginStorageComparableValue = (value: unknown): string | number | boolean | null => {
if (value instanceof Date) return value.getTime();
if (typeof value === "number" || typeof value === "string" || typeof value === "boolean") {
return value;
}
if (value == null) return null;
return JSON.stringify(value);
};

const comparePluginStorageValues = (left: unknown, right: unknown): number => {
const leftValue = pluginStorageComparableValue(left);
const rightValue = pluginStorageComparableValue(right);
if (leftValue === rightValue) return 0;
if (leftValue === null) return -1;
if (rightValue === null) return 1;
if (typeof leftValue === "number" && typeof rightValue === "number") {
return leftValue - rightValue;
}
return String(leftValue).localeCompare(String(rightValue));
};

const pluginStorageDataField = (data: unknown, field: string): unknown =>
isPluginStorageRecord(data) ? data[field] : undefined;

const matchesPluginStorageWhereValue = (actual: unknown, expected: unknown): boolean => {
if (!isPluginStorageWhereFilter(expected)) return Object.is(actual, expected);

if ("eq" in expected && !Object.is(actual, expected.eq)) return false;
if ("in" in expected) {
const values = expected.in;
if (!Array.isArray(values) || !values.some((value) => Object.is(actual, value))) return false;
}
if ("gt" in expected && !(comparePluginStorageValues(actual, expected.gt) > 0)) return false;
if ("gte" in expected && !(comparePluginStorageValues(actual, expected.gte) >= 0)) return false;
if ("lt" in expected && !(comparePluginStorageValues(actual, expected.lt) < 0)) return false;
if ("lte" in expected && !(comparePluginStorageValues(actual, expected.lte) <= 0)) return false;

return true;
};

const rowMatchesPluginStorageWhere = (
row: CoreRow<"plugin_storage">,
where: Readonly<Record<string, unknown>> | undefined,
): boolean => {
if (!where) return true;
return Object.entries(where).every(([field, expected]) =>
matchesPluginStorageWhereValue(pluginStorageDataField(row.data, field), expected),
);
};

const makePluginStorageFacade = (input: {
readonly core: ReturnType<typeof makeCoreDb>;
readonly pluginId: string;
Expand Down Expand Up @@ -828,7 +935,158 @@ const makePluginStorageFacade = (input: {
.pipe(Effect.map((rows) => sortByScopePrecedence(rows)[0] ?? null))
.pipe(Effect.map((row) => (row ? pluginStorageEntryFromRow<T>(row) : null)));

const queryCollection = <TDefinition extends PluginStorageCollectionDefinition>(
definition: TDefinition,
queryInput?: PluginStorageCollectionQueryInput<TDefinition>,
) =>
Effect.gen(function* () {
const validationError = pluginStorageQueryValidationError(
definition,
queryInput as
| PluginStorageCollectionQueryInput<PluginStorageCollectionDefinition>
| undefined,
);
if (validationError) return yield* validationError;

const rows = yield* input.core.findMany("plugin_storage", {
where: whereFor(definition.name),
});
const filtered = sortByScopePrecedence(rows)
.filter((row) =>
queryInput?.keyPrefix === undefined ? true : row.key.startsWith(queryInput.keyPrefix),
)
.filter((row) =>
rowMatchesPluginStorageWhere(
row,
queryInput?.where as Readonly<Record<string, unknown>> | undefined,
),
);

const sorted =
queryInput?.orderBy && queryInput.orderBy.length > 0
? [...filtered].sort((left, right) => {
for (const order of queryInput.orderBy ?? []) {
const direction = order.direction === "desc" ? -1 : 1;
const compared =
comparePluginStorageValues(
pluginStorageDataField(left.data, order.field),
pluginStorageDataField(right.data, order.field),
) * direction;
if (compared !== 0) return compared;
}
return (
input.scopeIds.indexOf(left.scope_id) - input.scopeIds.indexOf(right.scope_id) ||
left.key.localeCompare(right.key)
);
})
: filtered;

const offset = queryInput?.offset ?? 0;
const limited =
queryInput?.limit === undefined
? sorted.slice(offset)
: sorted.slice(offset, offset + queryInput.limit);
return limited.map((row) =>
pluginStorageEntryFromRow<PluginStorageCollectionData<TDefinition>>(row),
);
});

return {
collection: (definition) => ({
get: (storageInput) =>
getVisible(definition.name, storageInput.key) as Effect.Effect<
PluginStorageEntry<PluginStorageCollectionData<typeof definition>> | null,
StorageFailure
>,
getAtScope: (storageInput) =>
input.core
.findFirst("plugin_storage", {
where: byScopedId(
storageInput.scope,
pluginStorageId({
pluginId: input.pluginId,
collection: definition.name,
key: storageInput.key,
}),
),
})
.pipe(
Effect.map((row) =>
row
? pluginStorageEntryFromRow<PluginStorageCollectionData<typeof definition>>(row)
: null,
),
),
list: (storageInput) =>
queryCollection(definition, {
keyPrefix: storageInput?.keyPrefix,
}),
put: (storageInput) =>
Effect.gen(function* () {
if (!input.scopeIds.includes(storageInput.scope)) {
return yield* new StorageError({
message: `Unknown plugin storage target scope: ${storageInput.scope}`,
cause: undefined,
});
}
const row = yield* input.core.findFirst("plugin_storage", {
where: byScopedId(
storageInput.scope,
pluginStorageId({
pluginId: input.pluginId,
collection: definition.name,
key: storageInput.key,
}),
),
});
if (row) {
const now = new Date();
yield* input.core.updateMany("plugin_storage", {
where: byScopedId(storageInput.scope, row.id),
set: {
data: storageInput.data,
updated_at: now,
},
});
return pluginStorageEntryFromRow({
...row,
data: storageInput.data,
updated_at: now,
});
}

const now = new Date();
const created = yield* input.core.create("plugin_storage", {
id: pluginStorageId({
pluginId: input.pluginId,
collection: definition.name,
key: storageInput.key,
}),
scope_id: storageInput.scope,
plugin_id: input.pluginId,
collection: definition.name,
key: storageInput.key,
data: storageInput.data,
created_at: now,
updated_at: now,
});
return pluginStorageEntryFromRow(created);
}),
query: (queryInput) => queryCollection(definition, queryInput),
count: (queryInput) =>
queryCollection(definition, queryInput).pipe(Effect.map((entries) => entries.length)),
remove: (storageInput) =>
input.core.deleteMany("plugin_storage", {
where: byScopedId(
storageInput.scope,
pluginStorageId({
pluginId: input.pluginId,
collection: definition.name,
key: storageInput.key,
}),
),
}),
}),
get: (storageInput) => getVisible(storageInput.collection, storageInput.key),
getAtScope: (storageInput) =>
input.core
Expand Down
40 changes: 31 additions & 9 deletions packages/core/sdk/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,37 @@ export {
makeInMemoryBlobStore,
} from "./blob";

// Plugin storage
export {
definePluginStorageCollection,
pluginStorageId,
type PluginStorageCollectionDefinition,
type PluginStorageCollectionFacade,
type PluginStorageCollectionIndexedField,
type PluginStorageCollectionKeyInput,
type PluginStorageCollectionListInput,
type PluginStorageCollectionOrderBy,
type PluginStorageCollectionPutInput,
type PluginStorageCollectionQueryInput,
type PluginStorageCollectionScopedKeyInput,
type PluginStorageCollectionWhere,
type PluginStorageConfig,
type PluginStorageEntry,
type PluginStorageFacade,
type PluginStorageIndexField,
type PluginStorageIndexSpec,
type PluginStorageKeyInput,
type PluginStorageListInput,
type PluginStoragePutInput,
type PluginStorageRuntimeCollectionDefinition,
type PluginStorageRuntimeIndexSpec,
type PluginStorageSchema,
type PluginStorageSchemaType,
type PluginStorageScopedKeyInput,
type PluginStorageWhereFilter,
type PluginStorageWhereValue,
} from "./plugin-storage";

// OAuth 2.1
export {
type OAuthService,
Expand Down Expand Up @@ -320,15 +351,6 @@ export {
definePlugin,
tool,
} from "./plugin";
export {
pluginStorageId,
type PluginStorageEntry,
type PluginStorageFacade,
type PluginStorageKeyInput,
type PluginStorageListInput,
type PluginStoragePutInput,
type PluginStorageScopedKeyInput,
} from "./plugin-storage";

// Executor
export {
Expand Down
Loading
Loading