Execute oxfmt (#1166)
Ran pnpm oxfmt . This should get the web repo aligned so that we can better enforce oxfmt going forward.
This commit is contained in:
@@ -14,8 +14,13 @@ describe("SqlocalDraftRepository", () => {
|
||||
});
|
||||
|
||||
it("save then load returns the same text", async () => {
|
||||
await repo.save({ kind: "channel", channel: ChannelNumber.Primary }, "hello");
|
||||
expect(await repo.load({ kind: "channel", channel: ChannelNumber.Primary })).toBe("hello");
|
||||
await repo.save(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
"hello",
|
||||
);
|
||||
expect(
|
||||
await repo.load({ kind: "channel", channel: ChannelNumber.Primary }),
|
||||
).toBe("hello");
|
||||
});
|
||||
|
||||
it("save with empty text deletes the row", async () => {
|
||||
@@ -25,9 +30,14 @@ describe("SqlocalDraftRepository", () => {
|
||||
});
|
||||
|
||||
it("clear removes the row", async () => {
|
||||
await repo.save({ kind: "channel", channel: ChannelNumber.Channel1 }, "draft");
|
||||
await repo.save(
|
||||
{ kind: "channel", channel: ChannelNumber.Channel1 },
|
||||
"draft",
|
||||
);
|
||||
await repo.clear({ kind: "channel", channel: ChannelNumber.Channel1 });
|
||||
expect(await repo.load({ kind: "channel", channel: ChannelNumber.Channel1 })).toBe("");
|
||||
expect(
|
||||
await repo.load({ kind: "channel", channel: ChannelNumber.Channel1 }),
|
||||
).toBe("");
|
||||
});
|
||||
|
||||
it("upsert overwrites prior text without throwing", async () => {
|
||||
@@ -38,10 +48,20 @@ describe("SqlocalDraftRepository", () => {
|
||||
|
||||
it("scoped per device_id", async () => {
|
||||
const repoB = new SqlocalDraftRepository(db, { deviceId: 2 });
|
||||
await repo.save({ kind: "channel", channel: ChannelNumber.Primary }, "from-1");
|
||||
await repoB.save({ kind: "channel", channel: ChannelNumber.Primary }, "from-2");
|
||||
expect(await repo.load({ kind: "channel", channel: ChannelNumber.Primary })).toBe("from-1");
|
||||
expect(await repoB.load({ kind: "channel", channel: ChannelNumber.Primary })).toBe("from-2");
|
||||
await repo.save(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
"from-1",
|
||||
);
|
||||
await repoB.save(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
"from-2",
|
||||
);
|
||||
expect(
|
||||
await repo.load({ kind: "channel", channel: ChannelNumber.Primary }),
|
||||
).toBe("from-1");
|
||||
expect(
|
||||
await repoB.load({ kind: "channel", channel: ChannelNumber.Primary }),
|
||||
).toBe("from-2");
|
||||
});
|
||||
|
||||
it("loadAll returns all drafts for the device", async () => {
|
||||
|
||||
@@ -65,17 +65,24 @@ export class SqlocalDraftRepository implements DraftRepository {
|
||||
);
|
||||
}
|
||||
|
||||
async loadAll(): Promise<ReadonlyArray<{ key: ConversationKey; text: string }>> {
|
||||
async loadAll(): Promise<
|
||||
ReadonlyArray<{ key: ConversationKey; text: string }>
|
||||
> {
|
||||
const rows = await this.db
|
||||
.select({ conversationKey: drafts.conversationKey, text: drafts.text })
|
||||
.from(drafts)
|
||||
.where(eq(drafts.deviceId, this.deviceId));
|
||||
return rows.map((r) => ({ key: parseKey(r.conversationKey), text: r.text }));
|
||||
return rows.map((r) => ({
|
||||
key: parseKey(r.conversationKey),
|
||||
text: r.text,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
function parseKey(s: string): ConversationKey {
|
||||
if (s.startsWith("channel:")) return { kind: "channel", channel: Number(s.slice(8)) };
|
||||
if (s.startsWith("direct:")) return { kind: "direct", peer: Number(s.slice(7)) };
|
||||
if (s.startsWith("channel:"))
|
||||
return { kind: "channel", channel: Number(s.slice(8)) };
|
||||
if (s.startsWith("direct:"))
|
||||
return { kind: "direct", peer: Number(s.slice(7)) };
|
||||
throw new Error(`Unknown conversation key format: ${s}`);
|
||||
}
|
||||
|
||||
@@ -28,12 +28,20 @@ describe("SqlocalMessageRepository (sql.js test driver)", () => {
|
||||
|
||||
it("loadRecent returns the tail in chronological order", async () => {
|
||||
await repo.appendBatch([msg(1, 1000), msg(2, 2000), msg(3, 3000)]);
|
||||
const out = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 2);
|
||||
const out = await repo.loadRecent(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
2,
|
||||
);
|
||||
expect(out.map((m) => m.id)).toEqual([2, 3]);
|
||||
});
|
||||
|
||||
it("loadBefore paginates older messages", async () => {
|
||||
await repo.appendBatch([msg(1, 1000), msg(2, 2000), msg(3, 3000), msg(4, 4000)]);
|
||||
await repo.appendBatch([
|
||||
msg(1, 1000),
|
||||
msg(2, 2000),
|
||||
msg(3, 3000),
|
||||
msg(4, 4000),
|
||||
]);
|
||||
const out = await repo.loadBefore(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
new Date(3000),
|
||||
@@ -45,22 +53,39 @@ describe("SqlocalMessageRepository (sql.js test driver)", () => {
|
||||
it("updateState mutates the matching row", async () => {
|
||||
await repo.append(msg(42, 1000));
|
||||
await repo.updateState(42, MessageState.Failed);
|
||||
const [found] = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 1);
|
||||
const [found] = await repo.loadRecent(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
1,
|
||||
);
|
||||
expect(found?.state).toBe(MessageState.Failed);
|
||||
});
|
||||
|
||||
it("prune enforces maxPerBucket", async () => {
|
||||
await repo.appendBatch([msg(1, 1000), msg(2, 2000), msg(3, 3000), msg(4, 4000)]);
|
||||
await repo.appendBatch([
|
||||
msg(1, 1000),
|
||||
msg(2, 2000),
|
||||
msg(3, 3000),
|
||||
msg(4, 4000),
|
||||
]);
|
||||
await repo.prune({ maxPerBucket: 2 });
|
||||
const out = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 10);
|
||||
const out = await repo.loadRecent(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
10,
|
||||
);
|
||||
expect(out.map((m) => m.id)).toEqual([3, 4]);
|
||||
});
|
||||
|
||||
it("prune enforces olderThanMs", async () => {
|
||||
const now = Date.now();
|
||||
await repo.appendBatch([msg(1, now - 1000 * 60 * 60 * 24 * 10), msg(2, now)]);
|
||||
await repo.appendBatch([
|
||||
msg(1, now - 1000 * 60 * 60 * 24 * 10),
|
||||
msg(2, now),
|
||||
]);
|
||||
await repo.prune({ olderThanMs: 1000 * 60 * 60 * 24 });
|
||||
const out = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 10);
|
||||
const out = await repo.loadRecent(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
10,
|
||||
);
|
||||
expect(out.map((m) => m.id)).toEqual([2]);
|
||||
});
|
||||
|
||||
@@ -68,8 +93,14 @@ describe("SqlocalMessageRepository (sql.js test driver)", () => {
|
||||
const repoB = new SqlocalMessageRepository(db, { deviceId: 2 });
|
||||
await repo.append(msg(1, 1000, "from-1"));
|
||||
await repoB.append(msg(2, 2000, "from-2"));
|
||||
const a = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 10);
|
||||
const b = await repoB.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 10);
|
||||
const a = await repo.loadRecent(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
10,
|
||||
);
|
||||
const b = await repoB.loadRecent(
|
||||
{ kind: "channel", channel: ChannelNumber.Primary },
|
||||
10,
|
||||
);
|
||||
expect(a.map((m) => m.text)).toEqual(["from-1"]);
|
||||
expect(b.map((m) => m.text)).toEqual(["from-2"]);
|
||||
});
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import type { ConversationKey, Message, MessageRepository, RetentionPolicy } from "@meshtastic/sdk";
|
||||
import type {
|
||||
ConversationKey,
|
||||
Message,
|
||||
MessageRepository,
|
||||
RetentionPolicy,
|
||||
} from "@meshtastic/sdk";
|
||||
import { conversationKeyString, MessageState } from "@meshtastic/sdk";
|
||||
import { and, desc, eq, lt, sql } from "drizzle-orm";
|
||||
import { type MultiTabCoordinator } from "../coordination/MultiTabCoordinator.ts";
|
||||
@@ -33,7 +38,11 @@ export class SqlocalMessageRepository implements MessageRepository {
|
||||
return rows.reverse().map(rowToMessage);
|
||||
}
|
||||
|
||||
async loadBefore(key: ConversationKey, cursor: Date, limit: number): Promise<Message[]> {
|
||||
async loadBefore(
|
||||
key: ConversationKey,
|
||||
cursor: Date,
|
||||
limit: number,
|
||||
): Promise<Message[]> {
|
||||
const rows = await this.db
|
||||
.select()
|
||||
.from(messages)
|
||||
@@ -66,7 +75,12 @@ export class SqlocalMessageRepository implements MessageRepository {
|
||||
const cutoff = Date.now() - policy.olderThanMs;
|
||||
await this.db
|
||||
.delete(messages)
|
||||
.where(and(eq(messages.deviceId, this.deviceId), lt(messages.rxTime, cutoff))!);
|
||||
.where(
|
||||
and(
|
||||
eq(messages.deviceId, this.deviceId),
|
||||
lt(messages.rxTime, cutoff),
|
||||
)!,
|
||||
);
|
||||
}
|
||||
if (policy.maxPerBucket !== undefined) {
|
||||
// Keep the newest N per (device, conversation_key). SQLite has no
|
||||
|
||||
@@ -20,7 +20,11 @@ describe("MultiTabCoordinator broadcast", () => {
|
||||
});
|
||||
a.broadcast({ kind: "messages-changed", deviceId: 1, key: "channel:0" });
|
||||
const event = await received;
|
||||
expect(event).toEqual({ kind: "messages-changed", deviceId: 1, key: "channel:0" });
|
||||
expect(event).toEqual({
|
||||
kind: "messages-changed",
|
||||
deviceId: 1,
|
||||
key: "channel:0",
|
||||
});
|
||||
} finally {
|
||||
a.close();
|
||||
b.close();
|
||||
|
||||
@@ -12,7 +12,10 @@
|
||||
* so the chat slice in another tab can re-query after a write.
|
||||
*/
|
||||
|
||||
export type ChangeKind = "messages-changed" | "nodes-changed" | "telemetry-changed";
|
||||
export type ChangeKind =
|
||||
| "messages-changed"
|
||||
| "nodes-changed"
|
||||
| "telemetry-changed";
|
||||
|
||||
export interface ChangeEvent {
|
||||
kind: ChangeKind;
|
||||
@@ -63,10 +66,14 @@ export class MultiTabCoordinator {
|
||||
if (typeof navigator === "undefined" || !navigator.locks) {
|
||||
return handler();
|
||||
}
|
||||
const result = await navigator.locks.request(resource, options ?? {}, async (lock) => {
|
||||
if (lock === null) return undefined;
|
||||
return handler();
|
||||
});
|
||||
const result = await navigator.locks.request(
|
||||
resource,
|
||||
options ?? {},
|
||||
async (lock) => {
|
||||
if (lock === null) return undefined;
|
||||
return handler();
|
||||
},
|
||||
);
|
||||
return result as T;
|
||||
}
|
||||
|
||||
|
||||
@@ -16,23 +16,32 @@ export type SqlocalDb = BaseSQLiteDatabase<"async", unknown, typeof schema>;
|
||||
* returns a Drizzle client typed against the schema. One instance per origin —
|
||||
* sqlocal serializes writes via Web Locks under the hood.
|
||||
*/
|
||||
export async function createSqlocalDb(options: CreateSqlocalDbOptions = {}): Promise<SqlocalDb> {
|
||||
export async function createSqlocalDb(
|
||||
options: CreateSqlocalDbOptions = {},
|
||||
): Promise<SqlocalDb> {
|
||||
const databasePath = options.databasePath ?? "meshtastic.db";
|
||||
const { driver, batchDriver, sql } = new SQLocalDrizzle({ databasePath });
|
||||
const db = drizzle(driver, batchDriver, { schema, casing: "snake_case" }) as SqlocalDb;
|
||||
const db = drizzle(driver, batchDriver, {
|
||||
schema,
|
||||
casing: "snake_case",
|
||||
}) as SqlocalDb;
|
||||
|
||||
await applyMigrations(sql);
|
||||
|
||||
return db;
|
||||
}
|
||||
|
||||
type RawSql = (query: TemplateStringsArray, ...values: ReadonlyArray<unknown>) => Promise<unknown>;
|
||||
type RawSql = (
|
||||
query: TemplateStringsArray,
|
||||
...values: ReadonlyArray<unknown>
|
||||
) => Promise<unknown>;
|
||||
|
||||
async function applyMigrations(sql: RawSql): Promise<void> {
|
||||
await sql`CREATE TABLE IF NOT EXISTS _schema (version INTEGER PRIMARY KEY)`;
|
||||
const rows = (await sql`SELECT version FROM _schema ORDER BY version DESC LIMIT 1`) as Array<{
|
||||
version: number;
|
||||
}>;
|
||||
const rows =
|
||||
(await sql`SELECT version FROM _schema ORDER BY version DESC LIMIT 1`) as Array<{
|
||||
version: number;
|
||||
}>;
|
||||
const current = rows[0]?.version ?? 0;
|
||||
for (const migration of MIGRATIONS) {
|
||||
if (migration.version <= current) continue;
|
||||
|
||||
@@ -24,7 +24,10 @@ export class SqlocalNodesRepository implements NodesRepository {
|
||||
}
|
||||
|
||||
async loadAll(): Promise<Node[]> {
|
||||
const rows = await this.db.select().from(nodes).where(eq(nodes.deviceId, this.deviceId));
|
||||
const rows = await this.db
|
||||
.select()
|
||||
.from(nodes)
|
||||
.where(eq(nodes.deviceId, this.deviceId));
|
||||
return rows.map(rowToNode);
|
||||
}
|
||||
|
||||
@@ -96,7 +99,10 @@ function rowToNode(row: NodeRow): Node {
|
||||
? fromBinary(Protobuf.Mesh.PositionSchema, base64Decode(row.positionJson))
|
||||
: undefined,
|
||||
deviceMetrics: row.metricsJson
|
||||
? fromBinary(Protobuf.Telemetry.DeviceMetricsSchema, base64Decode(row.metricsJson))
|
||||
? fromBinary(
|
||||
Protobuf.Telemetry.DeviceMetricsSchema,
|
||||
base64Decode(row.metricsJson),
|
||||
)
|
||||
: undefined,
|
||||
lastHeard: row.lastHeard ?? undefined,
|
||||
snr: row.snr ?? undefined,
|
||||
@@ -113,12 +119,16 @@ function nodeToRow(deviceId: number, node: Node): NodeRow {
|
||||
snr: node.snr ?? null,
|
||||
isFavorite: node.isFavorite,
|
||||
isIgnored: node.isIgnored,
|
||||
userJson: node.user ? base64Encode(toBinary(Protobuf.Mesh.UserSchema, node.user)) : null,
|
||||
userJson: node.user
|
||||
? base64Encode(toBinary(Protobuf.Mesh.UserSchema, node.user))
|
||||
: null,
|
||||
positionJson: node.position
|
||||
? base64Encode(toBinary(Protobuf.Mesh.PositionSchema, node.position))
|
||||
: null,
|
||||
metricsJson: node.deviceMetrics
|
||||
? base64Encode(toBinary(Protobuf.Telemetry.DeviceMetricsSchema, node.deviceMetrics))
|
||||
? base64Encode(
|
||||
toBinary(Protobuf.Telemetry.DeviceMetricsSchema, node.deviceMetrics),
|
||||
)
|
||||
: null,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -22,7 +22,11 @@ export const messages = sqliteTable(
|
||||
},
|
||||
(t) => ({
|
||||
pk: index("messages_pk").on(t.deviceId, t.id),
|
||||
convRxTime: index("idx_messages_conv_rxtime").on(t.deviceId, t.conversationKey, t.rxTime),
|
||||
convRxTime: index("idx_messages_conv_rxtime").on(
|
||||
t.deviceId,
|
||||
t.conversationKey,
|
||||
t.rxTime,
|
||||
),
|
||||
pending: index("idx_messages_pending").on(t.deviceId, t.state),
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { integer, primaryKey, sqliteTable, text } from "drizzle-orm/sqlite-core";
|
||||
import {
|
||||
integer,
|
||||
primaryKey,
|
||||
sqliteTable,
|
||||
text,
|
||||
} from "drizzle-orm/sqlite-core";
|
||||
|
||||
export const drafts = sqliteTable(
|
||||
"drafts",
|
||||
|
||||
@@ -11,14 +11,21 @@ describe("MIGRATIONS", () => {
|
||||
it("first migration creates messages, nodes, telemetry, _schema", async () => {
|
||||
const db = await freshSqlite();
|
||||
for (const stmt of MIGRATIONS[0]!.sql) db.run(stmt);
|
||||
db.run("INSERT INTO _schema (version) VALUES (?)", [MIGRATIONS[0]!.version]);
|
||||
db.run("INSERT INTO _schema (version) VALUES (?)", [
|
||||
MIGRATIONS[0]!.version,
|
||||
]);
|
||||
|
||||
const tables = db
|
||||
.exec("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")[0]
|
||||
.exec(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name",
|
||||
)[0]
|
||||
?.values.flat() as string[];
|
||||
expect(tables).toEqual(expect.arrayContaining(["_schema", "messages", "nodes", "telemetry"]));
|
||||
expect(tables).toEqual(
|
||||
expect.arrayContaining(["_schema", "messages", "nodes", "telemetry"]),
|
||||
);
|
||||
|
||||
const version = db.exec("SELECT MAX(version) FROM _schema")[0]?.values[0]?.[0];
|
||||
const version = db.exec("SELECT MAX(version) FROM _schema")[0]
|
||||
?.values[0]?.[0];
|
||||
expect(version).toBe(MIGRATIONS[0]!.version);
|
||||
});
|
||||
|
||||
@@ -31,7 +38,11 @@ describe("MIGRATIONS", () => {
|
||||
)[0]
|
||||
?.values.flat() as string[];
|
||||
expect(indexes).toEqual(
|
||||
expect.arrayContaining(["idx_messages_conv_rxtime", "idx_messages_pending", "messages_pk"]),
|
||||
expect.arrayContaining([
|
||||
"idx_messages_conv_rxtime",
|
||||
"idx_messages_pending",
|
||||
"messages_pk",
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -1,4 +1,10 @@
|
||||
import { index, integer, primaryKey, sqliteTable, text } from "drizzle-orm/sqlite-core";
|
||||
import {
|
||||
index,
|
||||
integer,
|
||||
primaryKey,
|
||||
sqliteTable,
|
||||
text,
|
||||
} from "drizzle-orm/sqlite-core";
|
||||
|
||||
/**
|
||||
* Snapshot of the device's NodeDB. Position / metrics / user are stored as
|
||||
@@ -12,8 +18,12 @@ export const nodes = sqliteTable(
|
||||
num: integer("num").notNull(),
|
||||
lastHeard: integer("last_heard"),
|
||||
snr: integer("snr"),
|
||||
isFavorite: integer("is_favorite", { mode: "boolean" }).notNull().default(false),
|
||||
isIgnored: integer("is_ignored", { mode: "boolean" }).notNull().default(false),
|
||||
isFavorite: integer("is_favorite", { mode: "boolean" })
|
||||
.notNull()
|
||||
.default(false),
|
||||
isIgnored: integer("is_ignored", { mode: "boolean" })
|
||||
.notNull()
|
||||
.default(false),
|
||||
userJson: text("user_json"),
|
||||
positionJson: text("position_json"),
|
||||
metricsJson: text("metrics_json"),
|
||||
|
||||
@@ -11,7 +11,9 @@ function reading(nodeNum: number, ms: number, battery = 80): TelemetryReading {
|
||||
nodeNum,
|
||||
time: new Date(ms),
|
||||
kind: "deviceMetrics",
|
||||
value: create(Protobuf.Telemetry.DeviceMetricsSchema, { batteryLevel: battery }),
|
||||
value: create(Protobuf.Telemetry.DeviceMetricsSchema, {
|
||||
batteryLevel: battery,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -31,7 +33,9 @@ describe("SqlocalTelemetryRepository", () => {
|
||||
|
||||
const out = await repo.loadRecent(100, 10);
|
||||
expect(out.map((r) => r.time.getTime())).toEqual([1000, 2000, 3000]);
|
||||
expect((out[2]?.value as { batteryLevel?: number } | undefined)?.batteryLevel).toBe(70);
|
||||
expect(
|
||||
(out[2]?.value as { batteryLevel?: number } | undefined)?.batteryLevel,
|
||||
).toBe(70);
|
||||
});
|
||||
|
||||
it("loadBefore returns rows older than the cursor in ascending order", async () => {
|
||||
@@ -90,7 +94,9 @@ describe("SqlocalTelemetryRepository", () => {
|
||||
value,
|
||||
});
|
||||
const out = await repo.loadRecent(9, 1);
|
||||
const v = out[0]?.value as { batteryLevel?: number; voltage?: number } | undefined;
|
||||
const v = out[0]?.value as
|
||||
| { batteryLevel?: number; voltage?: number }
|
||||
| undefined;
|
||||
expect(v?.batteryLevel).toBe(73);
|
||||
expect(v?.voltage).toBeCloseTo(4.1);
|
||||
});
|
||||
|
||||
@@ -34,17 +34,29 @@ export class SqlocalTelemetryRepository implements TelemetryRepository {
|
||||
this.deviceId = options.deviceId;
|
||||
}
|
||||
|
||||
async loadRecent(nodeNum: number, limit: number): Promise<TelemetryReading[]> {
|
||||
async loadRecent(
|
||||
nodeNum: number,
|
||||
limit: number,
|
||||
): Promise<TelemetryReading[]> {
|
||||
const rows = await this.db
|
||||
.select()
|
||||
.from(telemetry)
|
||||
.where(and(eq(telemetry.deviceId, this.deviceId), eq(telemetry.nodeNum, nodeNum))!)
|
||||
.where(
|
||||
and(
|
||||
eq(telemetry.deviceId, this.deviceId),
|
||||
eq(telemetry.nodeNum, nodeNum),
|
||||
)!,
|
||||
)
|
||||
.orderBy(desc(telemetry.ts))
|
||||
.limit(limit);
|
||||
return rows.map(rowToReading).reverse();
|
||||
}
|
||||
|
||||
async loadBefore(nodeNum: number, cursor: Date, limit: number): Promise<TelemetryReading[]> {
|
||||
async loadBefore(
|
||||
nodeNum: number,
|
||||
cursor: Date,
|
||||
limit: number,
|
||||
): Promise<TelemetryReading[]> {
|
||||
const rows = await this.db
|
||||
.select()
|
||||
.from(telemetry)
|
||||
@@ -78,7 +90,9 @@ export class SqlocalTelemetryRepository implements TelemetryRepository {
|
||||
const cutoff = Date.now() - policy.olderThanMs;
|
||||
await this.db
|
||||
.delete(telemetry)
|
||||
.where(and(eq(telemetry.deviceId, this.deviceId), lt(telemetry.ts, cutoff))!);
|
||||
.where(
|
||||
and(eq(telemetry.deviceId, this.deviceId), lt(telemetry.ts, cutoff))!,
|
||||
);
|
||||
}
|
||||
if (policy.maxPerNode !== undefined) {
|
||||
const max = policy.maxPerNode;
|
||||
@@ -95,7 +109,12 @@ export class SqlocalTelemetryRepository implements TelemetryRepository {
|
||||
const cutoffRows = await this.db
|
||||
.select({ ts: telemetry.ts })
|
||||
.from(telemetry)
|
||||
.where(and(eq(telemetry.deviceId, this.deviceId), eq(telemetry.nodeNum, row.nodeNum))!)
|
||||
.where(
|
||||
and(
|
||||
eq(telemetry.deviceId, this.deviceId),
|
||||
eq(telemetry.nodeNum, row.nodeNum),
|
||||
)!,
|
||||
)
|
||||
.orderBy(desc(telemetry.ts))
|
||||
.limit(1)
|
||||
.offset(max - 1);
|
||||
@@ -119,11 +138,18 @@ export class SqlocalTelemetryRepository implements TelemetryRepository {
|
||||
async clearNode(nodeNum: number): Promise<void> {
|
||||
await this.db
|
||||
.delete(telemetry)
|
||||
.where(and(eq(telemetry.deviceId, this.deviceId), eq(telemetry.nodeNum, nodeNum))!);
|
||||
.where(
|
||||
and(
|
||||
eq(telemetry.deviceId, this.deviceId),
|
||||
eq(telemetry.nodeNum, nodeNum),
|
||||
)!,
|
||||
);
|
||||
}
|
||||
|
||||
async clear(): Promise<void> {
|
||||
await this.db.delete(telemetry).where(eq(telemetry.deviceId, this.deviceId));
|
||||
await this.db
|
||||
.delete(telemetry)
|
||||
.where(eq(telemetry.deviceId, this.deviceId));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,7 +161,10 @@ interface TelemetryRow {
|
||||
payloadJson: string;
|
||||
}
|
||||
|
||||
function readingToRow(deviceId: number, reading: TelemetryReading): TelemetryRow | null {
|
||||
function readingToRow(
|
||||
deviceId: number,
|
||||
reading: TelemetryReading,
|
||||
): TelemetryRow | null {
|
||||
if (!reading.kind) return null;
|
||||
const schema = SCHEMA_FOR_KIND[reading.kind as KnownKind];
|
||||
if (!schema) return null;
|
||||
|
||||
@@ -17,8 +17,13 @@ export async function createMemoryDb(): Promise<SqlocalDb> {
|
||||
for (const statement of migration.sql) {
|
||||
sqlite.run(statement);
|
||||
}
|
||||
sqlite.run("INSERT OR IGNORE INTO _schema (version) VALUES (?)", [migration.version]);
|
||||
sqlite.run("INSERT OR IGNORE INTO _schema (version) VALUES (?)", [
|
||||
migration.version,
|
||||
]);
|
||||
}
|
||||
|
||||
return drizzle(sqlite, { schema, casing: "snake_case" }) as unknown as SqlocalDb;
|
||||
return drizzle(sqlite, {
|
||||
schema,
|
||||
casing: "snake_case",
|
||||
}) as unknown as SqlocalDb;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user