neil/silverbullet
1
0
Fork 0
Code Activity

Complete redo of content indexing and querying (#517)

Browse Source 
Complete redo of data store
Introduces live queries and live templates
This commit is contained in:
Zef Hemel
2023-10-03 14:16:33 +02:00
committed by GitHub
parent 7af98e7c7b
commit 0313565610
200 changed files with 4675 additions and 4363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugos/lib/datastore.test.ts
+30 -18
View File
@@ -6,15 +6,17 @@ import { KvPrimitives } from "./kv_primitives.ts";
import { assertEquals } from "https://deno.land/std@0.165.0/testing/asserts.ts";
async function test(db: KvPrimitives) {
const dataStore = new DataStore(db);
await dataStore.set(["user", "peter"], { name: "Peter" });
await dataStore.set(["user", "hank"], { name: "Hank" });
let results = await dataStore.query({
const datastore = new DataStore(db, ["ds"], {
count: (arr: any[]) => arr.length,
});
await datastore.set(["user", "peter"], { name: "Peter" });
await datastore.set(["user", "hank"], { name: "Hank" });
let results = await datastore.query({
prefix: ["user"],
filter: ["=", "name", "Peter"],
filter: ["=", ["attr", "name"], ["string", "Peter"]],
});
assertEquals(results, [{ key: ["user", "peter"], value: { name: "Peter" } }]);
await dataStore.batchSet([
await datastore.batchSet<any>([
{ key: ["kv", "name"], value: "Zef" },
{ key: ["kv", "data"], value: new Uint8Array([1, 2, 3]) },
{
@@ -29,32 +31,42 @@ async function test(db: KvPrimitives) {
},
},
]);
assertEquals(await dataStore.get(["kv", "name"]), "Zef");
assertEquals(await dataStore.get(["kv", "data"]), new Uint8Array([1, 2, 3]));
results = await dataStore.query({
assertEquals(await datastore.get(["kv", "name"]), "Zef");
assertEquals(await datastore.get(["kv", "data"]), new Uint8Array([1, 2, 3]));
results = await datastore.query({
prefix: ["kv"],
filter: ["=", "", "Zef"],
filter: ["=~", ["attr", ""], ["regexp", "Z.f", "i"]],
});
assertEquals(results, [{ key: ["kv", "name"], value: "Zef" }]);
results = await dataStore.query({
results = await datastore.query({
prefix: ["kv"],
filter: ["and", ["=", "parents", "John"], [
filter: ["and", ["=", ["attr", "parents"], ["string", "John"]], [
"=",
"address.city",
"San Francisco",
["attr", ["attr", "address"], "city"],
["string", "San Francisco"],
]],
select: ["name"],
select: [
{ name: "parents" },
{
name: "name",
expr: ["+", ["attr", "name"], ["string", "!"]],
},
{
name: "parentCount",
expr: ["call", "count", [["attr", "parents"]]],
},
],
});
assertEquals(results.length, 1);
assertEquals(results[0], {
key: ["kv", "complicated"],
value: { name: "Frank" },
value: { name: "Frank!", parentCount: 2, parents: ["John", "Jane"] },
});
}
Deno.test("Test Deno KV DataStore", async () => {
const tmpFile = await Deno.makeTempFile();
const db = new DenoKvPrimitives(tmpFile);
await db.init();
const db = new DenoKvPrimitives(await Deno.openKv(tmpFile));
await test(db);
db.close();
await Deno.remove(tmpFile);
plugos/lib/datastore.ts
+80 -147
View File
@@ -1,183 +1,116 @@
import { KvKey, KvPrimitives } from "./kv_primitives.ts";
export type { KvKey };
export type KvValue = any;
export type KV = {
key: KvKey;
value: KvValue;
};
export type KvOrderBy = {
attribute: string;
desc: boolean;
};
export type KvQuery = {
prefix: KvKey;
filter?: KvQueryFilter;
orderBy?: KvOrderBy[];
limit?: number;
select?: string[];
};
export type KvQueryFilter =
| ["=", string, any]
| ["!=", string, any]
| ["=~", string, RegExp]
| ["!=~", string, RegExp]
| ["prefix", string, string]
| ["<", string, any]
| ["<=", string, any]
| [">", string, any]
| [">=", string, any]
| ["in", string, any[]]
| ["and", KvQueryFilter, KvQueryFilter]
| ["or", KvQueryFilter, KvQueryFilter];
function filterKvQuery(kvQuery: KvQueryFilter, obj: KvValue): boolean {
const [op, op1, op2] = kvQuery;
if (op === "and") {
return filterKvQuery(op1, obj) &&
filterKvQuery(op2, obj);
} else if (op === "or") {
return filterKvQuery(op1, obj) || filterKvQuery(op2, obj);
}
// Look up the value of the attribute, supporting nested attributes via `attr.attr2.attr3`, and empty attribute value signifies the root object
let attributeVal = obj;
for (const part of op1.split(".")) {
if (!part) {
continue;
}
if (attributeVal === undefined) {
return false;
}
attributeVal = attributeVal[part];
}
// And apply the operator
switch (op) {
case "=": {
if (Array.isArray(attributeVal) && !Array.isArray(op2)) {
// Record property is an array, and value is a scalar: find the value in the array
if (attributeVal.includes(op2)) {
return true;
}
} else if (Array.isArray(attributeVal) && Array.isArray(obj)) {
// Record property is an array, and value is an array: find the value in the array
if (attributeVal.some((v) => obj.includes(v))) {
return true;
}
}
return attributeVal === op2;
}
case "!=":
return attributeVal !== op2;
case "=~":
return op2.test(attributeVal);
case "!=~":
return !op2.test(attributeVal);
case "prefix":
return attributeVal.startsWith(op2);
case "<":
return attributeVal < op2;
case "<=":
return attributeVal <= op2;
case ">":
return attributeVal > op2;
case ">=":
return attributeVal >= op2;
case "in":
return op2.includes(attributeVal);
default:
throw new Error(`Unupported operator: ${op}`);
}
}
import { applyQueryNoFilterKV, evalQueryExpression } from "$sb/lib/query.ts";
import { FunctionMap, KV, KvKey, KvQuery } from "$sb/types.ts";
import { builtinFunctions } from "$sb/lib/builtin_query_functions.ts";
import { KvPrimitives } from "./kv_primitives.ts";
/**
* This is the data store class you'll actually want to use, wrapping the primitives
* in a more user-friendly way
*/
export class DataStore {
constructor(private kv: KvPrimitives) {
constructor(
private kv: KvPrimitives,
private prefix: KvKey = [],
private functionMap: FunctionMap = builtinFunctions,
) {
}
async get(key: KvKey): Promise<KvValue> {
return (await this.kv.batchGet([key]))[0];
prefixed(prefix: KvKey): DataStore {
return new DataStore(
this.kv,
[...this.prefix, ...prefix],
this.functionMap,
);
}
batchGet(keys: KvKey[]): Promise<KvValue[]> {
return this.kv.batchGet(keys);
async get<T = any>(key: KvKey): Promise<T | null> {
return (await this.batchGet([key]))[0];
}
set(key: KvKey, value: KvValue): Promise<void> {
return this.kv.batchSet([{ key, value }]);
batchGet<T = any>(keys: KvKey[]): Promise<(T | null)[]> {
return this.kv.batchGet(keys.map((key) => this.applyPrefix(key)));
}
batchSet(entries: KV[]): Promise<void> {
return this.kv.batchSet(entries);
set(key: KvKey, value: any): Promise<void> {
return this.batchSet([{ key, value }]);
}
batchSet<T = any>(entries: KV<T>[]): Promise<void> {
const allKeyStrings = new Set<string>();
const uniqueEntries: KV[] = [];
for (const { key, value } of entries) {
const keyString = JSON.stringify(key);
if (allKeyStrings.has(keyString)) {
console.warn(`Duplicate key ${keyString} in batchSet, skipping`);
} else {
allKeyStrings.add(keyString);
uniqueEntries.push({ key: this.applyPrefix(key), value });
}
}
return this.kv.batchSet(uniqueEntries);
}
delete(key: KvKey): Promise<void> {
return this.kv.batchDelete([key]);
return this.batchDelete([key]);
}
batchDelete(keys: KvKey[]): Promise<void> {
return this.kv.batchDelete(keys);
return this.kv.batchDelete(keys.map((key) => this.applyPrefix(key)));
}
async query(query: KvQuery): Promise<KV[]> {
const results: KV[] = [];
async query<T = any>(query: KvQuery): Promise<KV<T>[]> {
const results: KV<T>[] = [];
let itemCount = 0;
// Accumuliate results
for await (const entry of this.kv.query({ prefix: query.prefix })) {
// Accumulate results
let limit = Infinity;
const prefixedQuery: KvQuery = {
...query,
prefix: query.prefix ? this.applyPrefix(query.prefix) : undefined,
};
if (query.limit) {
limit = evalQueryExpression(query.limit, {}, this.functionMap);
}
for await (
const entry of this.kv.query(prefixedQuery)
) {
// Filter
if (query.filter && !filterKvQuery(query.filter, entry.value)) {
if (
query.filter &&
!evalQueryExpression(query.filter, entry.value, this.functionMap)
) {
continue;
}
results.push(entry);
itemCount++;
// Stop when the limit has been reached
if (itemCount === query.limit) {
if (itemCount === limit) {
break;
}
}
// Order by
if (query.orderBy) {
results.sort((a, b) => {
const aVal = a.value;
const bVal = b.value;
for (const { attribute, desc } of query.orderBy!) {
if (
aVal[attribute] < bVal[attribute] || aVal[attribute] === undefined
) {
return desc ? 1 : -1;
}
if (
aVal[attribute] > bVal[attribute] || bVal[attribute] === undefined
) {
return desc ? -1 : 1;
}
}
// Consider them equal. This helps with comparing arrays (like tags)
return 0;
});
}
// Apply order by, limit, and select
return applyQueryNoFilterKV(prefixedQuery, results, this.functionMap).map((
{ key, value },
) => ({ key: this.stripPrefix(key), value }));
}
if (query.select) {
for (let i = 0; i < results.length; i++) {
const rec = results[i].value;
const newRec: any = {};
for (const k of query.select) {
newRec[k] = rec[k];
}
results[i].value = newRec;
}
async queryDelete(query: KvQuery): Promise<void> {
const keys: KvKey[] = [];
for (
const { key } of await this.query({
...query,
prefix: query.prefix ? this.applyPrefix(query.prefix) : undefined,
})
) {
keys.push(key);
}
return results;
return this.batchDelete(keys);
}
private applyPrefix(key: KvKey): KvKey {
return [...this.prefix, ...(key ? key : [])];
}
private stripPrefix(key: KvKey): KvKey {
return key.slice(this.prefix.length);
}
}
plugos/lib/deno_kv_primitives.test.ts
+1 -2
View File
@@ -3,8 +3,7 @@ import { allTests } from "./kv_primitives.test.ts";
Deno.test("Test Deno KV Primitives", async () => {
const tmpFile = await Deno.makeTempFile();
const db = new DenoKvPrimitives(tmpFile);
await db.init();
const db = new DenoKvPrimitives(await Deno.openKv(tmpFile));
await allTests(db);
db.close();
await Deno.remove(tmpFile);
plugos/lib/deno_kv_primitives.ts
+5 -8
View File
@@ -1,15 +1,12 @@
/// <reference lib="deno.unstable" />
import { KV, KvKey, KvPrimitives, KvQueryOptions } from "./kv_primitives.ts";
const kvBatchSize = 10;
import { KV, KvKey } from "$sb/types.ts";
import { KvPrimitives, KvQueryOptions } from "./kv_primitives.ts";
const kvBatchSize = 100;
export class DenoKvPrimitives implements KvPrimitives {
db!: Deno.Kv;
constructor(private path?: string) {
}
async init() {
this.db = await Deno.openKv(this.path);
constructor(private db: Deno.Kv) {
}
async batchGet(keys: KvKey[]): Promise<any[]> {
plugos/lib/indexeddb_kv_primitives.ts
+9 -8
View File
@@ -1,32 +1,33 @@
import { KV, KvKey, KvPrimitives, KvQueryOptions } from "./kv_primitives.ts";
import { KV, KvKey } from "$sb/types.ts";
import { KvPrimitives, KvQueryOptions } from "./kv_primitives.ts";
import { IDBPDatabase, openDB } from "https://esm.sh/idb@7.1.1/with-async-ittr";
const sep = "\0";
const objectStoreName = "data";
export class IndexedDBKvPrimitives implements KvPrimitives {
db!: IDBPDatabase<any>;
constructor(
private dbName: string,
private objectStoreName: string = "data",
) {
}
async init() {
this.db = await openDB(this.dbName, 1, {
upgrade: (db) => {
db.createObjectStore(this.objectStoreName);
db.createObjectStore(objectStoreName);
},
});
}
batchGet(keys: KvKey[]): Promise<any[]> {
const tx = this.db.transaction(this.objectStoreName, "readonly");
const tx = this.db.transaction(objectStoreName, "readonly");
return Promise.all(keys.map((key) => tx.store.get(this.buildKey(key))));
}
async batchSet(entries: KV[]): Promise<void> {
const tx = this.db.transaction(this.objectStoreName, "readwrite");
const tx = this.db.transaction(objectStoreName, "readwrite");
await Promise.all([
...entries.map(({ key, value }) =>
tx.store.put(value, this.buildKey(key))
@@ -36,7 +37,7 @@ export class IndexedDBKvPrimitives implements KvPrimitives {
}
async batchDelete(keys: KvKey[]): Promise<void> {
const tx = this.db.transaction(this.objectStoreName, "readwrite");
const tx = this.db.transaction(objectStoreName, "readwrite");
await Promise.all([
...keys.map((key) => tx.store.delete(this.buildKey(key))),
tx.done,
@@ -44,12 +45,12 @@ export class IndexedDBKvPrimitives implements KvPrimitives {
}
async *query({ prefix }: KvQueryOptions): AsyncIterableIterator<KV> {
const tx = this.db.transaction(this.objectStoreName, "readonly");
const tx = this.db.transaction(objectStoreName, "readonly");
prefix = prefix || [];
for await (
const entry of tx.store.iterate(IDBKeyRange.bound(
this.buildKey([...prefix, ""]),
this.buildKey([...prefix, "\ufffe"]),
this.buildKey([...prefix, "\uffff"]),
))
) {
yield { key: this.extractKey(entry.key), value: entry.value };
plugos/lib/kv_primitives.test.ts
+2 -1
View File
@@ -1,5 +1,6 @@
import { KV, KvPrimitives } from "./kv_primitives.ts";
import { KvPrimitives } from "./kv_primitives.ts";
import { assertEquals } from "../../test_deps.ts";
import { KV } from "$sb/types.ts";
export async function allTests(db: KvPrimitives) {
await db.batchSet([
plugos/lib/kv_primitives.ts
+2 -8
View File
@@ -1,17 +1,11 @@
export type KvKey = string[];
export type KvValue = any;
export type KV = {
key: KvKey;
value: KvValue;
};
import { KV, KvKey } from "$sb/types.ts";
export type KvQueryOptions = {
prefix?: KvKey;
};
export interface KvPrimitives {
batchGet(keys: KvKey[]): Promise<(KvValue | undefined)[]>;
batchGet(keys: KvKey[]): Promise<(any | undefined)[]>;
batchSet(entries: KV[]): Promise<void>;
batchDelete(keys: KvKey[]): Promise<void>;
query(options: KvQueryOptions): AsyncIterableIterator<KV>;
plugos/lib/kv_store.deno_kv.test.ts
-58
View File
@@ -1,58 +0,0 @@
import { assertEquals } from "../../test_deps.ts";
import { DenoKVStore } from "./kv_store.deno_kv.ts";
Deno.test("Test KV index", async () => {
const tmpFile = await Deno.makeTempFile();
const denoKv = await Deno.openKv(tmpFile);
const kv = new DenoKVStore(denoKv);
await kv.set("name", "Peter");
assertEquals(await kv.get("name"), "Peter");
await kv.del("name");
assertEquals(await kv.has("name"), false);
await kv.batchSet([
{ key: "page:hello", value: "Hello" },
{ key: "page:hello2", value: "Hello 2" },
{ key: "page:hello3", value: "Hello 3" },
{ key: "something", value: "Something" },
{ key: "something1", value: "Something" },
{ key: "something2", value: "Something" },
{ key: "something3", value: "Something" },
{ key: "something4", value: "Something" },
{ key: "something5", value: "Something" },
{ key: "something6", value: "Something" },
{ key: "something7", value: "Something" },
{ key: "something8", value: "Something" },
{ key: "something9", value: "Something" },
{ key: "something10", value: "Something" },
{ key: "something11", value: "Something" },
{ key: "something12", value: "Something" },
{ key: "something13", value: "Something" },
{ key: "something14", value: "Something" },
{ key: "something15", value: "Something" },
{ key: "something16", value: "Something" },
{ key: "something17", value: "Something" },
{ key: "something18", value: "Something" },
{ key: "something19", value: "Something" },
]);
const results = await kv.queryPrefix("page:");
assertEquals(results.length, 3);
assertEquals(await kv.batchGet(["page:hello", "page:hello3"]), [
"Hello",
"Hello 3",
]);
await kv.deletePrefix("page:");
assertEquals(await kv.queryPrefix("page:"), []);
assertEquals((await kv.queryPrefix("")).length, 20);
await kv.deletePrefix("");
assertEquals(await kv.queryPrefix(""), []);
denoKv.close();
await Deno.remove(tmpFile);
});
plugos/lib/kv_store.deno_kv.ts
-112
View File
@@ -1,112 +0,0 @@
/// <reference lib="deno.unstable" />
import { KV, KVStore } from "./kv_store.ts";
const kvBatchSize = 10;
export class DenoKVStore implements KVStore {
constructor(private kv: Deno.Kv) {
}
del(key: string): Promise<void> {
return this.batchDelete([key]);
}
async deletePrefix(prefix: string): Promise<void> {
const allKeys: string[] = [];
for await (
const result of this.kv.list(
prefix
? {
start: [prefix],
end: [endRange(prefix)],
}
: { prefix: [] },
)
) {
allKeys.push(result.key[0] as string);
}
return this.batchDelete(allKeys);
}
deleteAll(): Promise<void> {
return this.deletePrefix("");
}
set(key: string, value: any): Promise<void> {
return this.batchSet([{ key, value }]);
}
async batchSet(kvs: KV[]): Promise<void> {
// Split into batches of kvBatchSize
const batches: KV[][] = [];
for (let i = 0; i < kvs.length; i += kvBatchSize) {
batches.push(kvs.slice(i, i + kvBatchSize));
}
for (const batch of batches) {
let batchOp = this.kv.atomic();
for (const { key, value } of batch) {
batchOp = batchOp.set([key], value);
}
const res = await batchOp.commit();
if (!res.ok) {
throw res;
}
}
}
async batchDelete(keys: string[]): Promise<void> {
const batches: string[][] = [];
for (let i = 0; i < keys.length; i += kvBatchSize) {
batches.push(keys.slice(i, i + kvBatchSize));
}
for (const batch of batches) {
let batchOp = this.kv.atomic();
for (const key of batch) {
batchOp = batchOp.delete([key]);
}
const res = await batchOp.commit();
if (!res.ok) {
throw res;
}
}
}
async batchGet(keys: string[]): Promise<any[]> {
const results: any[] = [];
const batches: Deno.KvKey[][] = [];
for (let i = 0; i < keys.length; i += kvBatchSize) {
batches.push(keys.slice(i, i + kvBatchSize).map((k) => [k]));
}
for (const batch of batches) {
const res = await this.kv.getMany(batch);
results.push(...res.map((r) => r.value));
}
return results;
}
async get(key: string): Promise<any> {
return (await this.kv.get([key])).value;
}
async has(key: string): Promise<boolean> {
return (await this.kv.get([key])).value !== null;
}
async queryPrefix(keyPrefix: string): Promise<{ key: string; value: any }[]> {
const results: { key: string; value: any }[] = [];
for await (
const result of this.kv.list(
keyPrefix
? {
start: [keyPrefix],
end: [endRange(keyPrefix)],
}
: { prefix: [] },
)
) {
results.push({
key: result.key[0] as string,
value: result.value as any,
});
}
return results;
}
}
function endRange(prefix: string) {
const lastChar = prefix[prefix.length - 1];
const nextLastChar = String.fromCharCode(lastChar.charCodeAt(0) + 1);
return prefix.slice(0, -1) + nextLastChar;
}
plugos/lib/kv_store.dexie.ts
-82
View File
@@ -1,82 +0,0 @@
import Dexie, { Table } from "dexie";
import type { KV, KVStore } from "./kv_store.ts";
export class DexieKVStore implements KVStore {
db: Dexie;
items: Table<KV, string>;
constructor(
dbName: string,
tableName: string,
indexedDB?: any,
IDBKeyRange?: any,
) {
this.db = new Dexie(dbName, {
indexedDB,
IDBKeyRange,
});
this.db.version(1).stores({
[tableName]: "key",
});
this.items = this.db.table<KV, string>(tableName);
}
async del(key: string) {
await this.items.delete(key);
}
async deletePrefix(prefix: string) {
await this.items.where("key").startsWith(prefix).delete();
}
async deleteAll() {
await this.items.clear();
}
async set(key: string, value: any) {
await this.items.put({
key,
value,
});
}
async batchSet(kvs: KV[]) {
await this.items.bulkPut(
kvs.map(({ key, value }) => ({
key,
value,
})),
);
}
async batchDelete(keys: string[]) {
await this.items.bulkDelete(keys);
}
async batchGet(
keys: string[],
): Promise<(any | undefined)[]> {
return (await this.items.bulkGet(keys)).map((result) => result?.value);
}
async get(key: string): Promise<any | null> {
const result = await this.items.get({ key });
return result ? result.value : null;
}
async has(key: string): Promise<boolean> {
return await this.items.get({
key,
}) !== undefined;
}
async queryPrefix(
keyPrefix: string,
): Promise<{ key: string; value: any }[]> {
const results = await this.items.where("key").startsWith(keyPrefix)
.toArray();
return results.map((result) => ({
key: result.key,
value: result.value,
}));
}
}
plugos/lib/kv_store.json_file.ts
+2 -17
View File
@@ -1,6 +1,6 @@
import { KV, KVStore } from "./kv_store.ts";
import { KV } from "$sb/types.ts";
export class JSONKVStore implements KVStore {
export class JSONKVStore {
private data: { [key: string]: any } = {};
async load(path: string) {
@@ -38,21 +38,6 @@ export class JSONKVStore implements KVStore {
this.data[key] = value;
return Promise.resolve();
}
batchSet(kvs: KV[]): Promise<void> {
for (const kv of kvs) {
this.data[kv.key] = kv.value;
}
return Promise.resolve();
}
batchDelete(keys: string[]): Promise<void> {
for (const key of keys) {
delete this.data[key];
}
return Promise.resolve();
}
batchGet(keys: string[]): Promise<any[]> {
return Promise.resolve(keys.map((key) => this.data[key]));
}
get(key: string): Promise<any> {
return Promise.resolve(this.data[key]);
}
plugos/lib/kv_store.ts
-59
View File
@@ -1,59 +0,0 @@
export type KV = {
key: string;
value: any;
};
/**
* An interface to any simple key-value store.
*/
export interface KVStore {
/**
* Deletes the value associated with a given key.
*/
del(key: string): Promise<void>;
/**
* Deletes all keys that start with a specific prefix.
*/
deletePrefix(prefix: string): Promise<void>;
/**
* Deletes all keys in the store.
*/
deleteAll(): Promise<void>;
/**
* Sets the value for a given key.
*/
set(key: string, value: any): Promise<void>;
/**
* Sets the values for a list of key-value pairs.
*/
batchSet(kvs: KV[]): Promise<void>;
/**
* Deletes a list of keys.
*/
batchDelete(keys: string[]): Promise<void>;
/**
* Gets the values for a list of keys.
*/
batchGet(keys: string[]): Promise<(any | undefined)[]>;
/**
* Gets the value for a given key.
*/
get(key: string): Promise<any | null>;
/**
* Checks whether a given key exists in the store.
*/
has(key: string): Promise<boolean>;
/**
* Gets all key-value pairs where the key starts with a specific prefix.
*/
queryPrefix(keyPrefix: string): Promise<{ key: string; value: any }[]>;
}
plugos/lib/mq.dexie.test.ts → plugos/lib/mq.datastore.test.ts
+17 -7
View File
@@ -1,10 +1,14 @@
import { IDBKeyRange, indexedDB } from "https://esm.sh/fake-indexeddb@4.0.2";
import { DexieMQ } from "./mq.dexie.ts";
import { DataStoreMQ } from "./mq.datastore.ts";
import { assertEquals } from "../../test_deps.ts";
import { sleep } from "$sb/lib/async.ts";
import { DenoKvPrimitives } from "./deno_kv_primitives.ts";
import { DataStore } from "./datastore.ts";
Deno.test("Dexie MQ", async () => {
const mq = new DexieMQ("test", indexedDB, IDBKeyRange);
Deno.test("DataStore MQ", async () => {
const tmpFile = await Deno.makeTempFile();
const db = new DenoKvPrimitives(await Deno.openKv(tmpFile));
const mq = new DataStoreMQ(new DataStore(db, ["mq"]));
await mq.send("test", "Hello World");
let messages = await mq.poll("test", 10);
assertEquals(messages.length, 1);
@@ -28,12 +32,15 @@ Deno.test("Dexie MQ", async () => {
let receivedMessage = false;
const unsubscribe = mq.subscribe("test123", {}, async (messages) => {
assertEquals(messages.length, 1);
await mq.ack("test123", messages[0].id);
receivedMessage = true;
console.log("RECEIVED TEH EMSSSAGE");
await mq.ack("test123", messages[0].id);
});
mq.send("test123", "Hello World");
await mq.send("test123", "Hello World");
console.log("After send");
// Give time to process the message
await sleep(1);
await sleep(10);
console.log("After sleep");
assertEquals(receivedMessage, true);
unsubscribe();
@@ -50,4 +57,7 @@ Deno.test("Dexie MQ", async () => {
assertEquals(await mq.fetchProcessingMessages(), []);
// Give time to close the db
await sleep(20);
db.close();
await Deno.remove(tmpFile);
});
plugos/lib/mq.datastore.ts
+276
View File
@@ -0,0 +1,276 @@
import { KV, MQMessage, MQStats, MQSubscribeOptions } from "$sb/types.ts";
import { MessageQueue } from "./mq.ts";
import { DataStore } from "./datastore.ts";
export type ProcessingMessage = MQMessage & {
ts: number;
};
const queuedPrefix = ["mq", "queued"];
const processingPrefix = ["mq", "processing"];
const dlqPrefix = ["mq", "dlq"];
export class DataStoreMQ implements MessageQueue {
// queue -> set of run() functions
localSubscriptions = new Map<string, Set<() => void>>();
constructor(
private ds: DataStore,
) {
}
// Internal sequencer for messages, only really necessary when batch sending tons of messages within a millisecond
seq = 0;
async batchSend(queue: string, bodies: any[]): Promise<void> {
const messages: KV<MQMessage>[] = bodies.map((body) => {
const id = `${Date.now()}-${String(++this.seq).padStart(6, "0")}`;
const key = [...queuedPrefix, queue, id];
return {
key,
value: { id, queue, body },
};
});
await this.ds.batchSet(messages);
// See if we can immediately process the message with a local subscription
const localSubscriptions = this.localSubscriptions.get(queue);
if (localSubscriptions) {
for (const run of localSubscriptions) {
run();
}
}
}
send(queue: string, body: any): Promise<void> {
return this.batchSend(queue, [body]);
}
async poll(queue: string, maxItems: number): Promise<MQMessage[]> {
// Note: this is not happening in a transactional way, so we may get duplicate message delivery
// Retrieve a batch of messages
const messages = await this.ds.query<MQMessage>({
prefix: [...queuedPrefix, queue],
limit: ["number", maxItems],
});
// Put them in the processing queue
await this.ds.batchSet(
messages.map((m) => ({
key: [...processingPrefix, queue, m.value.id],
value: {
...m.value,
ts: Date.now(),
},
})),
);
// Delete them from the queued queue
await this.ds.batchDelete(messages.map((m) => m.key));
// Return them
return messages.map((m) => m.value);
}
/**
* @param queue
* @param batchSize
* @param callback
* @returns a function to be called to unsubscribe
*/
subscribe(
queue: string,
options: MQSubscribeOptions,
callback: (messages: MQMessage[]) => Promise<void> | void,
): () => void {
let running = true;
let timeout: number | undefined;
const batchSize = options.batchSize || 1;
const run = async () => {
try {
if (!running) {
return;
}
const messages = await this.poll(queue, batchSize);
if (messages.length > 0) {
await callback(messages);
}
// If we got exactly the batch size, there might be more messages
if (messages.length === batchSize) {
await run();
}
if (timeout) {
clearTimeout(timeout);
}
timeout = setTimeout(run, options.pollInterval || 5000);
} catch (e: any) {
console.error("Error in MQ subscription handler", e);
}
};
// Register as a local subscription handler
const localSubscriptions = this.localSubscriptions.get(queue);
if (!localSubscriptions) {
this.localSubscriptions.set(queue, new Set([run]));
} else {
localSubscriptions.add(run);
}
// Run the first time (which will schedule subsequent polling intervals)
run();
// And return an unsubscribe function
return () => {
running = false;
if (timeout) {
clearTimeout(timeout);
}
// Remove the subscription from localSubscriptions
const queueSubscriptions = this.localSubscriptions.get(queue);
if (queueSubscriptions) {
queueSubscriptions.delete(run);
}
};
}
ack(queue: string, id: string) {
return this.batchAck(queue, [id]);
}
async batchAck(queue: string, ids: string[]) {
await this.ds.batchDelete(
ids.map((id) => [...processingPrefix, queue, id]),
);
}
async requeueTimeouts(
timeout: number,
maxRetries?: number,
disableDLQ?: boolean,
) {
const now = Date.now();
const messages = await this.ds.query<ProcessingMessage>({
prefix: processingPrefix,
filter: ["<", ["attr", "ts"], ["number", now - timeout]],
});
await this.ds.batchDelete(messages.map((m) => m.key));
const newMessages: KV<ProcessingMessage>[] = [];
for (const { value: m } of messages) {
const retries = (m.retries || 0) + 1;
if (maxRetries && retries > maxRetries) {
if (disableDLQ) {
console.warn(
"[mq]",
"Message exceeded max retries, flushing message",
m,
);
} else {
console.warn(
"[mq]",
"Message exceeded max retries, moving to DLQ",
m,
);
newMessages.push({
key: [...dlqPrefix, m.queue, m.id],
value: {
queue: m.queue,
id: m.id,
body: m.body,
ts: Date.now(),
retries,
},
});
}
} else {
console.info("[mq]", "Message ack timed out, requeueing", m);
newMessages.push({
key: [...queuedPrefix, m.queue, m.id],
value: {
...m,
retries,
},
});
}
}
await this.ds.batchSet(newMessages);
}
async fetchDLQMessages(): Promise<ProcessingMessage[]> {
return (await this.ds.query<ProcessingMessage>({ prefix: dlqPrefix })).map((
{ value },
) => value);
}
async fetchProcessingMessages(): Promise<ProcessingMessage[]> {
return (await this.ds.query<ProcessingMessage>({
prefix: processingPrefix,
})).map((
{ value },
) => value);
}
flushDLQ(): Promise<void> {
return this.ds.queryDelete({ prefix: dlqPrefix });
}
async getQueueStats(queue: string): Promise<MQStats> {
const queued =
(await (this.ds.query({ prefix: [...queuedPrefix, queue] }))).length;
const processing =
(await (this.ds.query({ prefix: [...processingPrefix, queue] }))).length;
const dlq =
(await (this.ds.query({ prefix: [...dlqPrefix, queue] }))).length;
return {
queued,
processing,
dlq,
};
}
async getAllQueueStats(): Promise<Record<string, MQStats>> {
const allStatus: Record<string, MQStats> = {};
for (
const { value: message } of await this.ds.query<MQMessage>({
prefix: queuedPrefix,
})
) {
if (!allStatus[message.queue]) {
allStatus[message.queue] = {
queued: 0,
processing: 0,
dlq: 0,
};
}
allStatus[message.queue].queued++;
}
for (
const { value: message } of await this.ds.query<MQMessage>({
prefix: processingPrefix,
})
) {
if (!allStatus[message.queue]) {
allStatus[message.queue] = {
queued: 0,
processing: 0,
dlq: 0,
};
}
allStatus[message.queue].processing++;
}
for (
const { value: message } of await this.ds.query<MQMessage>({
prefix: dlqPrefix,
})
) {
if (!allStatus[message.queue]) {
allStatus[message.queue] = {
queued: 0,
processing: 0,
dlq: 0,
};
}
allStatus[message.queue].dlq++;
}
return allStatus;
}
}
plugos/lib/mq.deno_kv.test.ts
-20
View File
@@ -1,20 +0,0 @@
import { sleep } from "$sb/lib/async.ts";
import { DenoKvMQ } from "./mq.deno_kv.ts";
Deno.test("Deno MQ", async () => {
const denoKv = await Deno.openKv("test.db");
const mq = new DenoKvMQ(denoKv);
const unsub = mq.subscribe("test", {}, (messages) => {
console.log("Received on test", messages);
});
const unsub2 = mq.subscribe("test2", {}, (messages) => {
console.log("Received on test2", messages);
});
await mq.send("test", "Hello World");
await mq.batchSend("test2", ["Hello World 2", "Hello World 3"]);
// Let's avoid a panic here
await sleep(20);
denoKv.close();
await Deno.remove("test.db");
});
plugos/lib/mq.deno_kv.ts
-93
View File
@@ -1,93 +0,0 @@
/// <reference lib="deno.unstable" />
import {
MQMessage,
MQStats,
MQSubscribeOptions,
} from "../../plug-api/types.ts";
import { MessageQueue } from "./mq.ts";
type QueuedMessage = [string, MQMessage];
export class DenoKvMQ implements MessageQueue {
listeners: Map<string, Set<(messages: MQMessage[]) => void | Promise<void>>> =
new Map();
constructor(private kv: Deno.Kv) {
kv.listenQueue(async (message: unknown) => {
const [queue, body] = message as QueuedMessage;
const listeners = this.listeners.get(queue);
if (!listeners) {
return;
}
for (const listener of listeners) {
await Promise.resolve(listener([{ id: "_dummyid", queue, body }]));
}
});
}
// Dummy implementation
getQueueStats(_queue: string): Promise<MQStats> {
return Promise.resolve({
queued: 0,
processing: 0,
dlq: 0,
});
}
// Dummy implementation
getAllQueueStats(): Promise<Record<string, MQStats>> {
return Promise.resolve({});
}
async batchSend(queue: string, bodies: any[]): Promise<void> {
for (const body of bodies) {
const result = await this.kv.enqueue([queue, body]);
if (!result.ok) {
throw result;
}
}
// const results = await Promise.all(
// bodies.map((body) => this.kv.enqueue([queue, body])),
// );
// for (const result of results) {
// if (!result.ok) {
// throw result;
// }
// }
}
async send(queue: string, body: any): Promise<void> {
const result = await this.kv.enqueue([queue, body]);
if (!result.ok) {
throw result;
}
}
subscribe(
queue: string,
_options: MQSubscribeOptions,
callback: (messages: MQMessage[]) => void | Promise<void>,
): () => void {
const listeners = this.listeners.get(queue);
if (!listeners) {
this.listeners.set(queue, new Set([callback]));
} else {
listeners.add(callback);
}
return () => {
const listeners = this.listeners.get(queue);
if (!listeners) {
return;
}
listeners.delete(callback);
};
}
ack(_queue: string, _id: string): Promise<void> {
// Doesn't apply to this implementation
return Promise.resolve();
}
batchAck(_queue: string, _ids: string[]): Promise<void> {
// Doesn't apply to this implementation
return Promise.resolve();
}
}
plugos/lib/mq.dexie.ts
-279
View File
@@ -1,279 +0,0 @@
import Dexie, { Table } from "dexie";
import { MQMessage, MQStats, MQSubscribeOptions } from "$sb/types.ts";
import { MessageQueue } from "./mq.ts";
export type ProcessingMessage = MQMessage & {
ts: number;
};
export class DexieMQ implements MessageQueue {
db: Dexie;
queued: Table<MQMessage, [string, string]>;
processing: Table<ProcessingMessage, [string, string]>;
dlq: Table<ProcessingMessage, [string, string]>;
// queue -> set of run() functions
localSubscriptions = new Map<string, Set<() => void>>();
constructor(
dbName: string,
indexedDB?: any,
IDBKeyRange?: any,
) {
this.db = new Dexie(dbName, {
indexedDB,
IDBKeyRange,
});
this.db.version(1).stores({
queued: "[queue+id], queue, id",
processing: "[queue+id], queue, id, ts",
dlq: "[queue+id], queue, id",
});
this.queued = this.db.table("queued");
this.processing = this.db.table("processing");
this.dlq = this.db.table("dlq");
}
// Internal sequencer for messages, only really necessary when batch sending tons of messages within a millisecond
seq = 0;
async batchSend(queue: string, bodies: any[]) {
const messages = bodies.map((body) => ({
id: `${Date.now()}-${String(++this.seq).padStart(6, "0")}`,
queue,
body,
}));
await this.queued.bulkAdd(messages);
// See if we can immediately process the message with a local subscription
const localSubscriptions = this.localSubscriptions.get(queue);
if (localSubscriptions) {
for (const run of localSubscriptions) {
run();
}
}
}
send(queue: string, body: any) {
return this.batchSend(queue, [body]);
}
poll(queue: string, maxItems: number): Promise<MQMessage[]> {
return this.db.transaction(
"rw",
[this.queued, this.processing],
async (tx) => {
const messages =
(await tx.table<MQMessage, [string, string]>("queued").where({
queue,
})
.sortBy("id")).slice(0, maxItems);
const ids: [string, string][] = messages.map((m) => [queue, m.id]);
await tx.table("queued").bulkDelete(ids);
await tx.table<ProcessingMessage, [string, string]>("processing")
.bulkPut(
messages.map((m) => ({
...m,
ts: Date.now(),
})),
);
return messages;
},
);
}
/**
* @param queue
* @param batchSize
* @param callback
* @returns a function to be called to unsubscribe
*/
subscribe(
queue: string,
options: MQSubscribeOptions,
callback: (messages: MQMessage[]) => Promise<void> | void,
): () => void {
let running = true;
let timeout: number | undefined;
const batchSize = options.batchSize || 1;
const run = async () => {
try {
if (!running) {
return;
}
const messages = await this.poll(queue, batchSize);
if (messages.length > 0) {
await callback(messages);
}
// If we got exactly the batch size, there might be more messages
if (messages.length === batchSize) {
await run();
}
if (timeout) {
clearTimeout(timeout);
}
timeout = setTimeout(run, options.pollInterval || 5000);
} catch (e: any) {
console.error("Error in MQ subscription handler", e);
}
};
// Register as a local subscription handler
const localSubscriptions = this.localSubscriptions.get(queue);
if (!localSubscriptions) {
this.localSubscriptions.set(queue, new Set([run]));
} else {
localSubscriptions.add(run);
}
// Run the first time (which will schedule subsequent polling intervals)
run();
// And return an unsubscribe function
return () => {
running = false;
if (timeout) {
clearTimeout(timeout);
}
// Remove the subscription from localSubscriptions
const queueSubscriptions = this.localSubscriptions.get(queue);
if (queueSubscriptions) {
queueSubscriptions.delete(run);
}
};
}
ack(queue: string, id: string) {
return this.batchAck(queue, [id]);
}
async batchAck(queue: string, ids: string[]) {
await this.processing.bulkDelete(ids.map((id) => [queue, id]));
}
async requeueTimeouts(
timeout: number,
maxRetries?: number,
disableDLQ?: boolean,
) {
const now = Date.now();
const messages = await this.processing.where("ts").below(now - timeout)
.toArray();
const ids: [string, string][] = messages.map((m) => [m.queue, m.id]);
await this.db.transaction(
"rw",
[this.queued, this.processing, this.dlq],
async (tx) => {
await tx.table("processing").bulkDelete(ids);
const requeuedMessages: ProcessingMessage[] = [];
const dlqMessages: ProcessingMessage[] = [];
for (const m of messages) {
const retries = (m.retries || 0) + 1;
if (maxRetries && retries > maxRetries) {
if (disableDLQ) {
console.warn(
"[mq]",
"Message exceeded max retries, flushing message",
m,
);
} else {
console.warn(
"[mq]",
"Message exceeded max retries, moving to DLQ",
m,
);
dlqMessages.push({
queue: m.queue,
id: m.id,
body: m.body,
ts: Date.now(),
retries,
});
}
} else {
console.info("[mq]", "Message ack timed out, requeueing", m);
requeuedMessages.push({
...m,
retries,
});
}
}
await tx.table("queued").bulkPut(requeuedMessages);
await tx.table("dlq").bulkPut(dlqMessages);
},
);
}
fetchDLQMessages(): Promise<ProcessingMessage[]> {
return this.dlq.toArray();
}
fetchProcessingMessages(): Promise<ProcessingMessage[]> {
return this.processing.toArray();
}
flushDLQ(): Promise<void> {
return this.dlq.clear();
}
getQueueStats(queue: string): Promise<MQStats> {
return this.db.transaction(
"r",
[this.queued, this.processing, this.dlq],
async (tx) => {
const queued = await tx.table("queued").where({ queue }).count();
const processing = await tx.table("processing").where({ queue })
.count();
const dlq = await tx.table("dlq").where({ queue }).count();
return {
queued,
processing,
dlq,
};
},
);
}
async getAllQueueStats(): Promise<Record<string, MQStats>> {
const allStatus: Record<string, MQStats> = {};
await this.db.transaction(
"r",
[this.queued, this.processing, this.dlq],
async (tx) => {
for (const item of await tx.table("queued").toArray()) {
if (!allStatus[item.queue]) {
allStatus[item.queue] = {
queued: 0,
processing: 0,
dlq: 0,
};
}
allStatus[item.queue].queued++;
}
for (const item of await tx.table("processing").toArray()) {
if (!allStatus[item.queue]) {
allStatus[item.queue] = {
queued: 0,
processing: 0,
dlq: 0,
};
}
allStatus[item.queue].processing++;
}
for (const item of await tx.table("dlq").toArray()) {
if (!allStatus[item.queue]) {
allStatus[item.queue] = {
queued: 0,
processing: 0,
dlq: 0,
};
}
allStatus[item.queue].dlq++;
}
},
);
return allStatus;
}
}