fix options; all analytics exists inside singular db now

packages/analytics-engine/README.md

@@ -10,7 +10,7 @@ fun, full-featured, fully-local simulator for Cloudflare Workers. See
 import { AnalyticsEngine } from "@miniflare/analytics-engine";
 import { createSQLiteDB } from "@miniflare/shared";
 
-const db = new AnalyticsEngine("TEST_BINDING", await createSQLiteDB(":memory:"));
+const db = new AnalyticsEngine("DATASET_NAME", await createSQLiteDB(":memory:"));
 
 await db.writeDataPoint({
  indexes: ["a3cd45"], // Sensor ID

packages/analytics-engine/src/analytics.ts

@@ -47,8 +47,8 @@ CREATE TABLE IF NOT EXISTS {{BINDING}} (
  blob20 TEXT
 );
 
-CREATE INDEX IF NOT EXISTS {{BINDING}}_index ON {{BINDING}} (dataset, timestamp);
-CREATE INDEX IF NOT EXISTS {{BINDING}}_index ON {{BINDING}} (dataset, index1, timestamp);
+CREATE UNIQUE INDEX IF NOT EXISTS {{BINDING}}_index ON {{BINDING}} (dataset, timestamp);
+CREATE UNIQUE INDEX IF NOT EXISTS {{BINDING}}_index ON {{BINDING}} (dataset, index1, timestamp);
 
 COMMIT;
 `;

packages/analytics-engine/src/engine.ts

@@ -54,59 +54,17 @@ interface DataPoint {
  blob20?: string | null;
 }
 
-type Doubles =
- | "double1"
- | "double2"
- | "double3"
- | "double4"
- | "double5"
- | "double6"
- | "double7"
- | "double8"
- | "double9"
- | "double10"
- | "double11"
- | "double12"
- | "double13"
- | "double14"
- | "double15"
- | "double16"
- | "double17"
- | "double18"
- | "double19"
- | "double20";
-
-type Blobs =
- | "blob1"
- | "blob2"
- | "blob3"
- | "blob4"
- | "blob5"
- | "blob6"
- | "blob7"
- | "blob8"
- | "blob9"
- | "blob10"
- | "blob11"
- | "blob12"
- | "blob13"
- | "blob14"
- | "blob15"
- | "blob16"
- | "blob17"
- | "blob18"
- | "blob19"
- | "blob20";
+export const kQuery = Symbol("kQuery");
 
 export class AnalyticsEngine {
- readonly #name: string;
+ readonly #dataset: string;
  readonly #db: SqliteDB;
  #decoder = new TextDecoder();
 
- constructor(name: string, db: SqliteDB) {
- this.#name = name;
+ constructor(dataset: string, db: SqliteDB) {
+ this.#dataset = dataset;
  this.#db = db;
- db.exec(analytics.replaceAll("{{BINDING}}", name));
+ db.exec(analytics.replaceAll("{{BINDING}}", dataset));
  buildSQLFunctions(db);
  }
 
@@ -146,30 +104,30 @@ export class AnalyticsEngine {
  }
  // prep insert
  const insertData: DataPoint = {
- dataset: this.#name,
+ dataset: this.#dataset,
  index1: indexes[0],
  };
  // prep doubles
  const doublesKeys: string[] = [];
  const doublesValues: string[] = [];
  doubles.forEach((double, i) => {
- const key = `double${i + 1}` as Doubles;
- insertData[key] = double;
+ const key = `double${i + 1}` as keyof DataPoint;
+ (insertData[key] as any) = double;
  doublesKeys.push(key);
  doublesValues.push(`@${key}`);
  });
  // prep blobs
  const blobsKeys: string[] = [];
  const blobsValues: string[] = [];
  _blobs.forEach((blob, i) => {
- const key = `blob${i + 1}` as Blobs;
- insertData[key] = blob;
+ const key = `blob${i + 1}` as keyof DataPoint;
+ (insertData[key] as any) = blob;
  blobsKeys.push(key);
  blobsValues.push(`@${key}`);
  });
 
  const input = prepare(
- `INSERT INTO ${this.#name} (dataset, index1${
+ `INSERT INTO ${this.#dataset} (dataset, index1${
  doublesKeys.length > 0 ? `, ${doublesKeys}` : ""
  }${
  blobsKeys.length > 0 ? `, ${blobsKeys}` : ""
@@ -181,6 +139,12 @@ export class AnalyticsEngine {
 
  insert.run(insertData);
  }
+
+ async [kQuery](input: string): Promise<any> {
+ const query = this.#db.prepare(prepare(input));
+
+ return query.get();
+ }
 }
 
 /** @internal */

packages/analytics-engine/src/plugin.ts

@@ -8,10 +8,13 @@ import {
  StorageFactory,
  resolveStoragePersist,
 } from "@miniflare/shared";
+import type { SqliteDB } from "@miniflare/shared";
 import { AnalyticsEngine } from "./engine";
 
+export type ProcessedAnalyticsEngine = Record<string, string>; // { [name]: dataset }
+
 export interface AnalyticsEngineOptions {
- analyticsEngines?: string[];
+ analyticsEngines?: ProcessedAnalyticsEngine;
  aePersist?: boolean | string;
 }
 
@@ -20,14 +23,25 @@ export class AnalyticsEnginePlugin
  implements AnalyticsEngineOptions
 {
  @Option({
- type: OptionType.ARRAY,
- name: "analyticsEngine",
- description: "Analytics Engine namespace to bind",
- logName: "Analytics Engine Namespaces",
- fromWrangler: ({ analytics_engines }) =>
- analytics_engines?.map(({ binding }) => binding),
+ type: OptionType.OBJECT,
+ typeFormat: "NAME=DATASET",
+ name: "ae",
+ alias: "a",
+ description: "Analytics Engine to bind",
+ logName: "Analytics Engine Names",
+ fromEntries: (entries) =>
+ Object.fromEntries(
+ entries.map(([name, datasetName]) => {
+ return [name, datasetName];
+ })
+ ),
+ fromWrangler: ({ bindings }) =>
+ bindings?.reduce((objects, { type, name, dataset }) => {
+ if (type === "analytics_engine") objects[name] = dataset;
+ return objects;
+ }, {} as ProcessedAnalyticsEngine),
  })
- analyticsEngines?: string[];
+ analyticsEngines?: ProcessedAnalyticsEngine;
 
  @Option({
  type: OptionType.BOOLEAN_STRING,
@@ -38,6 +52,8 @@ export class AnalyticsEnginePlugin
  aePersist?: boolean | string;
  readonly #persist?: boolean | string;
 
+ #db?: SqliteDB;
+
  constructor(ctx: PluginContext, options?: AnalyticsEngineOptions) {
  super(ctx);
  this.assignOptions(options);
@@ -46,17 +62,35 @@ export class AnalyticsEnginePlugin
 
  async getAnalyticsEngine(
  storageFactory: StorageFactory,
- dbName: string
+ name: string
  ): Promise<AnalyticsEngine> {
- const storage = storageFactory.storage(dbName, this.#persist);
- return new AnalyticsEngine(dbName, await storage.getSqliteDatabase());
+ const dataset = this.analyticsEngines?.[name];
+ if (dataset === undefined) {
+ throw new Error(`Analytics Engine "${name}" does not exist.`);
+ }
+ await this.#setup(storageFactory);
+ // @ts-expect-error: #setup already ensures #db exists.
+ return new AnalyticsEngine(dataset, this.#db);
  }
 
  async setup(storageFactory: StorageFactory): Promise<SetupResult> {
+ await this.#setup(storageFactory);
  const bindings: Context = {};
- for (const dbName of this.analyticsEngines ?? []) {
- bindings[dbName] = await this.getAnalyticsEngine(storageFactory, dbName);
+ for (const name of Object.keys(this.analyticsEngines ?? {})) {
+ bindings[name] = await this.getAnalyticsEngine(storageFactory, name);
  }
  return { bindings };
  }
+
+ async #setup(storageFactory: StorageFactory): Promise<void> {
+ if (this.#db === undefined) {
+ // grab storage
+ const storage = storageFactory.storage(
+ "__MINIFLARE_ANALYTICS_ENGINE_STORAGE__",
+ this.#persist
+ );
+ // setup db
+ this.#db = await storage.getSqliteDatabase();
+ }
+ }
 }

packages/analytics-engine/test/engine.spec.ts

@@ -14,7 +14,7 @@ const test = anyTest as TestInterface<Context>;
 test.beforeEach(async (t) => {
  const storage = new MemoryStorage(undefined, testClock);
  const db = new AnalyticsEngine(
- "TEST_BINDING",
+ "TEST_DATASET",
  await storage.getSqliteDatabase()
  );
  t.context = { storage, db };
@@ -32,12 +32,12 @@ test("Analytics Engine: Write a data point using indexes, blobs, and doubles.",
  });
 
  // grab data from sqliteDB
- const stmt = sqliteDB.prepare("SELECT * FROM TEST_BINDING WHERE index1 = ?");
+ const stmt = sqliteDB.prepare("SELECT * FROM TEST_DATASET WHERE index1 = ?");
  const res = stmt.get("t1");
  t.true(typeof res.timestamp === "string");
  delete res.timestamp;
  t.deepEqual(res, {
- dataset: "TEST_BINDING",
+ dataset: "TEST_DATASET",
  index1: "t1",
  _sample_interval: 1,
  blob1: "a",
@@ -91,12 +91,12 @@ test("Analytics Engine: Write a data point with no data provided.", async (t) =>
  await db.writeDataPoint({});
 
  // grab data from sqliteDB
- const stmt = sqliteDB.prepare("SELECT * FROM TEST_BINDING");
+ const stmt = sqliteDB.prepare("SELECT * FROM TEST_DATASET");
  const res = stmt.get();
  t.true(typeof res.timestamp === "string");
  delete res.timestamp;
  t.deepEqual(res, {
- dataset: "TEST_BINDING",
+ dataset: "TEST_DATASET",
  index1: null,
  _sample_interval: 1,
  blob1: null,
@@ -177,12 +177,12 @@ test("Analytics Engine: Write a data point filling indexes, blobs, and doubles."
  });
 
  // grab data from sqliteDB
- const stmt = sqliteDB.prepare("SELECT * FROM TEST_BINDING WHERE index1 = ?");
+ const stmt = sqliteDB.prepare("SELECT * FROM TEST_DATASET WHERE index1 = ?");
  const res = stmt.get("t1");
  t.true(typeof res.timestamp === "string");
  delete res.timestamp;
  t.deepEqual(res, {
- dataset: "TEST_BINDING",
+ dataset: "TEST_DATASET",
  index1: "t1",
  _sample_interval: 1,
  blob1: "a",
@@ -241,7 +241,7 @@ test("Analytics Engine: Store AB", async (t) => {
  });
 
  const stmt = sqliteDB.prepare(
- "SELECT blob1 FROM TEST_BINDING WHERE index1 = ?"
+ "SELECT blob1 FROM TEST_DATASET WHERE index1 = ?"
  );
  const res = stmt.get("t1");
  t.is(res.blob1, "test string");
@@ -263,7 +263,7 @@ test("Analytics Engine: Minimal example test.", async (t) => {
 
  // grab data from sqliteDB
  const stmt = sqliteDB.prepare(
- "SELECT blob1 AS city, SUM(_sample_interval * double1) / SUM(_sample_interval) AS avg_humidity FROM TEST_BINDING WHERE double1 > 0 GROUP BY city ORDER BY avg_humidity DESC LIMIT 10"
+ "SELECT blob1 AS city, SUM(_sample_interval * double1) / SUM(_sample_interval) AS avg_humidity FROM TEST_DATASET WHERE double1 > 0 GROUP BY city ORDER BY avg_humidity DESC LIMIT 10"
  );
  const res = stmt.get();
  delete res.timestamp;
@@ -274,7 +274,7 @@ test("Analytics Engine: Minimal example test.", async (t) => {
 
  // USING TIME SERIES DATA
  const stmt2 = sqliteDB.prepare(
- "SELECT intDiv(toUInt32(timestamp), 300) * 300 AS t, blob1 AS city, SUM(_sample_interval * double1) / SUM(_sample_interval) AS avg_humidity FROM TEST_BINDING WHERE timestamp >= NOW() AND double1 > 0 GROUP BY t, city ORDER BY t, avg_humidity DESC"
+ "SELECT intDiv(toUInt32(timestamp), 300) * 300 AS t, blob1 AS city, SUM(_sample_interval * double1) / SUM(_sample_interval) AS avg_humidity FROM TEST_DATASET WHERE timestamp >= NOW() AND double1 > 0 GROUP BY t, city ORDER BY t, avg_humidity DESC"
  );
  const res2 = stmt2.get();
  // console.log("res2", res2);