mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-03-30 15:55:49 +00:00
feat: optimize observations table schema (v3 migration)
- Replace observer_id TEXT (64-char hex) + observer_name TEXT with observer_idx INTEGER (FK to observers rowid) - Remove redundant hash TEXT and created_at TEXT columns from observations - Store timestamp as INTEGER epoch seconds instead of ISO text string - Auto-migrate old schema on startup: backup DB, migrate data, rebuild indexes, VACUUM - Migration is safe: backup first, abort on failure, schema_version marker prevents re-runs - Backward-compatible packets_v view: JOINs observers table, converts epoch→ISO for consumers - In-memory observer_id→rowid Map for fast lookups during ingestion - In-memory dedup Set with 5-min TTL to prevent duplicate INSERT attempts - packet-store.js: detect v3 schema and use appropriate JOIN query - Tests: 29 migration tests (old→new, idempotency, backup failure, ingestion, dedup) - Tests: 19 new v3 schema tests in test-db.js (columns, types, view compat, ingestion) Expected savings on 947K-row prod DB: - observer_id: 61 bytes → 4 bytes per row (57 bytes saved) - observer_name: ~15 bytes → 0 (resolved via JOIN) - hash: 16 bytes → 0 (redundant with transmission_id) - timestamp: 25 bytes → 4 bytes (21 bytes saved) - created_at: 25 bytes → 0 (redundant) - Dedup index: much smaller (integers vs text) - Estimated ~118 bytes saved per row = ~112MB total + massive index savings
This commit is contained in:
you
324
db.js
324
db.js
@@ -67,45 +67,195 @@ db.exec(`
|
||||
created_at TEXT DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS observations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
transmission_id INTEGER NOT NULL REFERENCES transmissions(id),
|
||||
hash TEXT NOT NULL,
|
||||
observer_id TEXT,
|
||||
observer_name TEXT,
|
||||
direction TEXT,
|
||||
snr REAL,
|
||||
rssi REAL,
|
||||
score INTEGER,
|
||||
path_json TEXT,
|
||||
timestamp TEXT NOT NULL,
|
||||
created_at TEXT DEFAULT (datetime('now'))
|
||||
CREATE TABLE IF NOT EXISTS schema_version (
|
||||
version INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_transmissions_hash ON transmissions(hash);
|
||||
CREATE INDEX IF NOT EXISTS idx_transmissions_first_seen ON transmissions(first_seen);
|
||||
CREATE INDEX IF NOT EXISTS idx_transmissions_payload_type ON transmissions(payload_type);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_hash ON observations(hash);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_transmission_id ON observations(transmission_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_observer_id ON observations(observer_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_timestamp ON observations(timestamp);
|
||||
DROP INDEX IF EXISTS idx_observations_dedup;
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_observations_dedup ON observations(hash, observer_id, COALESCE(path_json, ''));
|
||||
|
||||
-- Clean up legacy duplicates (same hash+observer+path, keep lowest id)
|
||||
DELETE FROM observations WHERE id NOT IN (
|
||||
SELECT MIN(id) FROM observations GROUP BY hash, observer_id, COALESCE(path_json, '')
|
||||
);
|
||||
|
||||
CREATE VIEW IF NOT EXISTS packets_v AS
|
||||
SELECT o.id, t.raw_hex, o.timestamp, o.observer_id, o.observer_name,
|
||||
o.direction, o.snr, o.rssi, o.score, t.hash, t.route_type,
|
||||
t.payload_type, t.payload_version, o.path_json, t.decoded_json,
|
||||
t.created_at
|
||||
FROM observations o
|
||||
JOIN transmissions t ON t.id = o.transmission_id;
|
||||
`);
|
||||
|
||||
// --- Determine schema version ---
|
||||
let schemaVersion = 0;
|
||||
try {
|
||||
const row = db.prepare('SELECT version FROM schema_version ORDER BY version DESC LIMIT 1').get();
|
||||
if (row) schemaVersion = row.version;
|
||||
} catch {}
|
||||
|
||||
// --- v3 migration: lean observations table ---
|
||||
function needsV3Migration() {
|
||||
if (schemaVersion >= 3) return false;
|
||||
// Check if observations table exists with old observer_id TEXT column
|
||||
const obsExists = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='observations'").get();
|
||||
if (!obsExists) return false;
|
||||
const cols = db.pragma('table_info(observations)').map(c => c.name);
|
||||
return cols.includes('observer_id');
|
||||
}
|
||||
|
||||
function runV3Migration() {
|
||||
const startTime = Date.now();
|
||||
console.log('[migration-v3] Starting observations table optimization...');
|
||||
|
||||
// a. Backup DB
|
||||
const backupPath = dbPath + '.pre-v3-backup';
|
||||
try {
|
||||
console.log(`[migration-v3] Backing up DB to ${backupPath}...`);
|
||||
fs.copyFileSync(dbPath, backupPath);
|
||||
console.log(`[migration-v3] Backup complete (${Date.now() - startTime}ms)`);
|
||||
} catch (e) {
|
||||
console.error(`[migration-v3] Backup failed, aborting migration: ${e.message}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
// b. Create lean table
|
||||
let stepStart = Date.now();
|
||||
db.exec(`
|
||||
CREATE TABLE observations_v3 (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
transmission_id INTEGER NOT NULL REFERENCES transmissions(id),
|
||||
observer_idx INTEGER,
|
||||
direction TEXT,
|
||||
snr REAL,
|
||||
rssi REAL,
|
||||
score INTEGER,
|
||||
path_json TEXT,
|
||||
timestamp INTEGER NOT NULL
|
||||
)
|
||||
`);
|
||||
console.log(`[migration-v3] Created observations_v3 table (${Date.now() - stepStart}ms)`);
|
||||
|
||||
// c. Migrate data
|
||||
stepStart = Date.now();
|
||||
const result = db.prepare(`
|
||||
INSERT INTO observations_v3 (id, transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp)
|
||||
SELECT o.id, o.transmission_id, obs.rowid, o.direction, o.snr, o.rssi, o.score, o.path_json,
|
||||
CAST(strftime('%s', o.timestamp) AS INTEGER)
|
||||
FROM observations o
|
||||
LEFT JOIN observers obs ON obs.id = o.observer_id
|
||||
`).run();
|
||||
console.log(`[migration-v3] Migrated ${result.changes} rows (${Date.now() - stepStart}ms)`);
|
||||
|
||||
// d. Drop old table
|
||||
stepStart = Date.now();
|
||||
db.exec('DROP TABLE observations');
|
||||
console.log(`[migration-v3] Dropped old observations table (${Date.now() - stepStart}ms)`);
|
||||
|
||||
// e. Rename
|
||||
stepStart = Date.now();
|
||||
db.exec('ALTER TABLE observations_v3 RENAME TO observations');
|
||||
console.log(`[migration-v3] Renamed observations_v3 → observations (${Date.now() - stepStart}ms)`);
|
||||
|
||||
// f. Create indexes
|
||||
stepStart = Date.now();
|
||||
db.exec(`
|
||||
CREATE INDEX idx_observations_transmission_id ON observations(transmission_id);
|
||||
CREATE INDEX idx_observations_observer_idx ON observations(observer_idx);
|
||||
CREATE INDEX idx_observations_timestamp ON observations(timestamp);
|
||||
CREATE UNIQUE INDEX idx_observations_dedup ON observations(transmission_id, observer_idx, COALESCE(path_json, ''));
|
||||
`);
|
||||
console.log(`[migration-v3] Created indexes (${Date.now() - stepStart}ms)`);
|
||||
|
||||
// g. Set schema version
|
||||
db.exec('DELETE FROM schema_version');
|
||||
db.prepare('INSERT INTO schema_version (version) VALUES (?)').run(3);
|
||||
schemaVersion = 3;
|
||||
|
||||
// h. Rebuild view (done below in common code)
|
||||
|
||||
// i. VACUUM
|
||||
stepStart = Date.now();
|
||||
db.exec('VACUUM');
|
||||
console.log(`[migration-v3] VACUUM complete (${Date.now() - stepStart}ms)`);
|
||||
|
||||
console.log(`[migration-v3] Migration complete! Total time: ${Date.now() - startTime}ms`);
|
||||
return true;
|
||||
} catch (e) {
|
||||
console.error(`[migration-v3] Migration failed: ${e.message}`);
|
||||
console.error('[migration-v3] Old data should still be intact if observations table was not yet dropped');
|
||||
// Try to clean up v3 table if it exists
|
||||
try { db.exec('DROP TABLE IF EXISTS observations_v3'); } catch {}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
const isV3 = schemaVersion >= 3;
|
||||
|
||||
if (!isV3 && needsV3Migration()) {
|
||||
runV3Migration();
|
||||
}
|
||||
|
||||
// If schema_version < 3 and no migration happened (fresh DB or migration skipped), create old-style table
|
||||
if (schemaVersion < 3) {
|
||||
const obsExists = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='observations'").get();
|
||||
if (!obsExists) {
|
||||
// Fresh DB — create v3 schema directly
|
||||
db.exec(`
|
||||
CREATE TABLE observations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
transmission_id INTEGER NOT NULL REFERENCES transmissions(id),
|
||||
observer_idx INTEGER,
|
||||
direction TEXT,
|
||||
snr REAL,
|
||||
rssi REAL,
|
||||
score INTEGER,
|
||||
path_json TEXT,
|
||||
timestamp INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX idx_observations_transmission_id ON observations(transmission_id);
|
||||
CREATE INDEX idx_observations_observer_idx ON observations(observer_idx);
|
||||
CREATE INDEX idx_observations_timestamp ON observations(timestamp);
|
||||
CREATE UNIQUE INDEX idx_observations_dedup ON observations(transmission_id, observer_idx, COALESCE(path_json, ''));
|
||||
`);
|
||||
db.exec('DELETE FROM schema_version');
|
||||
db.prepare('INSERT INTO schema_version (version) VALUES (?)').run(3);
|
||||
schemaVersion = 3;
|
||||
} else {
|
||||
// Old-style observations table exists but migration wasn't run (or failed)
|
||||
// Ensure indexes exist for old schema
|
||||
db.exec(`
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_hash ON observations(hash);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_transmission_id ON observations(transmission_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_observer_id ON observations(observer_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_timestamp ON observations(timestamp);
|
||||
`);
|
||||
// Dedup cleanup for old schema
|
||||
try {
|
||||
db.exec(`DROP INDEX IF EXISTS idx_observations_dedup`);
|
||||
db.exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_observations_dedup ON observations(hash, observer_id, COALESCE(path_json, ''))`);
|
||||
db.exec(`DELETE FROM observations WHERE id NOT IN (SELECT MIN(id) FROM observations GROUP BY hash, observer_id, COALESCE(path_json, ''))`);
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
|
||||
// --- Create/rebuild packets_v view ---
|
||||
db.exec('DROP VIEW IF EXISTS packets_v');
|
||||
if (schemaVersion >= 3) {
|
||||
db.exec(`
|
||||
CREATE VIEW packets_v AS
|
||||
SELECT o.id, t.raw_hex,
|
||||
datetime(o.timestamp, 'unixepoch') AS timestamp,
|
||||
obs.id AS observer_id, obs.name AS observer_name,
|
||||
o.direction, o.snr, o.rssi, o.score, t.hash, t.route_type,
|
||||
t.payload_type, t.payload_version, o.path_json, t.decoded_json,
|
||||
t.created_at
|
||||
FROM observations o
|
||||
JOIN transmissions t ON t.id = o.transmission_id
|
||||
LEFT JOIN observers obs ON obs.rowid = o.observer_idx
|
||||
`);
|
||||
} else {
|
||||
db.exec(`
|
||||
CREATE VIEW packets_v AS
|
||||
SELECT o.id, t.raw_hex, o.timestamp, o.observer_id, o.observer_name,
|
||||
o.direction, o.snr, o.rssi, o.score, t.hash, t.route_type,
|
||||
t.payload_type, t.payload_version, o.path_json, t.decoded_json,
|
||||
t.created_at
|
||||
FROM observations o
|
||||
JOIN transmissions t ON t.id = o.transmission_id
|
||||
`);
|
||||
}
|
||||
|
||||
// --- Migrations for existing DBs ---
|
||||
const observerCols = db.pragma('table_info(observers)').map(c => c.name);
|
||||
for (const col of ['model', 'firmware', 'client_version', 'radio', 'battery_mv', 'uptime_secs', 'noise_floor']) {
|
||||
@@ -183,24 +333,66 @@ const stmts = {
|
||||
countPackets: db.prepare(`SELECT COUNT(*) as count FROM observations`),
|
||||
countNodes: db.prepare(`SELECT COUNT(*) as count FROM nodes`),
|
||||
countObservers: db.prepare(`SELECT COUNT(*) as count FROM observers`),
|
||||
countRecentPackets: db.prepare(`SELECT COUNT(*) as count FROM observations WHERE timestamp > ?`),
|
||||
countRecentPackets: schemaVersion >= 3
|
||||
? db.prepare(`SELECT COUNT(*) as count FROM observations WHERE timestamp > CAST(strftime('%s', ?) AS INTEGER)`)
|
||||
: db.prepare(`SELECT COUNT(*) as count FROM observations WHERE timestamp > ?`),
|
||||
getTransmissionByHash: db.prepare(`SELECT id, first_seen FROM transmissions WHERE hash = ?`),
|
||||
insertTransmission: db.prepare(`
|
||||
INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json)
|
||||
VALUES (@raw_hex, @hash, @first_seen, @route_type, @payload_type, @payload_version, @decoded_json)
|
||||
`),
|
||||
updateTransmissionFirstSeen: db.prepare(`UPDATE transmissions SET first_seen = @first_seen WHERE id = @id`),
|
||||
insertObservation: db.prepare(`
|
||||
INSERT OR IGNORE INTO observations (transmission_id, hash, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp)
|
||||
VALUES (@transmission_id, @hash, @observer_id, @observer_name, @direction, @snr, @rssi, @score, @path_json, @timestamp)
|
||||
`),
|
||||
insertObservation: schemaVersion >= 3
|
||||
? db.prepare(`
|
||||
INSERT OR IGNORE INTO observations (transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp)
|
||||
VALUES (@transmission_id, @observer_idx, @direction, @snr, @rssi, @score, @path_json, @timestamp)
|
||||
`)
|
||||
: db.prepare(`
|
||||
INSERT OR IGNORE INTO observations (transmission_id, hash, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp)
|
||||
VALUES (@transmission_id, @hash, @observer_id, @observer_name, @direction, @snr, @rssi, @score, @path_json, @timestamp)
|
||||
`),
|
||||
getObserverRowid: db.prepare(`SELECT rowid FROM observers WHERE id = ?`),
|
||||
};
|
||||
|
||||
// --- In-memory observer map (observer_id text → rowid integer) ---
|
||||
const observerIdToRowid = new Map();
|
||||
if (schemaVersion >= 3) {
|
||||
const rows = db.prepare('SELECT id, rowid FROM observers').all();
|
||||
for (const r of rows) observerIdToRowid.set(r.id, r.rowid);
|
||||
}
|
||||
|
||||
// --- In-memory dedup set for v3 ---
|
||||
const dedupSet = new Map(); // key → timestamp (for cleanup)
|
||||
const DEDUP_TTL_MS = 5 * 60 * 1000; // 5 minutes
|
||||
|
||||
function cleanupDedupSet() {
|
||||
const cutoff = Date.now() - DEDUP_TTL_MS;
|
||||
for (const [key, ts] of dedupSet) {
|
||||
if (ts < cutoff) dedupSet.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
// Periodic cleanup every 60s
|
||||
setInterval(cleanupDedupSet, 60000).unref();
|
||||
|
||||
function resolveObserverIdx(observerId) {
|
||||
if (!observerId) return null;
|
||||
let rowid = observerIdToRowid.get(observerId);
|
||||
if (rowid !== undefined) return rowid;
|
||||
// Try DB lookup (observer may have been inserted elsewhere)
|
||||
const row = stmts.getObserverRowid.get(observerId);
|
||||
if (row) {
|
||||
observerIdToRowid.set(observerId, row.rowid);
|
||||
return row.rowid;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// --- Helper functions ---
|
||||
|
||||
function insertTransmission(data) {
|
||||
const hash = data.hash;
|
||||
if (!hash) return null; // Can't deduplicate without a hash
|
||||
if (!hash) return null;
|
||||
|
||||
const timestamp = data.timestamp || new Date().toISOString();
|
||||
let transmissionId;
|
||||
@@ -208,7 +400,6 @@ function insertTransmission(data) {
|
||||
const existing = stmts.getTransmissionByHash.get(hash);
|
||||
if (existing) {
|
||||
transmissionId = existing.id;
|
||||
// Update first_seen if this observation is earlier
|
||||
if (timestamp < existing.first_seen) {
|
||||
stmts.updateTransmissionFirstSeen.run({ id: transmissionId, first_seen: timestamp });
|
||||
}
|
||||
@@ -225,18 +416,42 @@ function insertTransmission(data) {
|
||||
transmissionId = result.lastInsertRowid;
|
||||
}
|
||||
|
||||
const obsResult = stmts.insertObservation.run({
|
||||
transmission_id: transmissionId,
|
||||
hash,
|
||||
observer_id: data.observer_id || null,
|
||||
observer_name: data.observer_name || null,
|
||||
direction: data.direction || null,
|
||||
snr: data.snr ?? null,
|
||||
rssi: data.rssi ?? null,
|
||||
score: data.score ?? null,
|
||||
path_json: data.path_json || null,
|
||||
timestamp,
|
||||
});
|
||||
let obsResult;
|
||||
if (schemaVersion >= 3) {
|
||||
const observerIdx = resolveObserverIdx(data.observer_id);
|
||||
const epochTs = typeof timestamp === 'number' ? timestamp : Math.floor(new Date(timestamp).getTime() / 1000);
|
||||
|
||||
// In-memory dedup check
|
||||
const dedupKey = `${transmissionId}|${observerIdx}|${data.path_json || ''}`;
|
||||
if (dedupSet.has(dedupKey)) {
|
||||
return { transmissionId, observationId: 0 };
|
||||
}
|
||||
|
||||
obsResult = stmts.insertObservation.run({
|
||||
transmission_id: transmissionId,
|
||||
observer_idx: observerIdx,
|
||||
direction: data.direction || null,
|
||||
snr: data.snr ?? null,
|
||||
rssi: data.rssi ?? null,
|
||||
score: data.score ?? null,
|
||||
path_json: data.path_json || null,
|
||||
timestamp: epochTs,
|
||||
});
|
||||
dedupSet.set(dedupKey, Date.now());
|
||||
} else {
|
||||
obsResult = stmts.insertObservation.run({
|
||||
transmission_id: transmissionId,
|
||||
hash,
|
||||
observer_id: data.observer_id || null,
|
||||
observer_name: data.observer_name || null,
|
||||
direction: data.direction || null,
|
||||
snr: data.snr ?? null,
|
||||
rssi: data.rssi ?? null,
|
||||
score: data.score ?? null,
|
||||
path_json: data.path_json || null,
|
||||
timestamp,
|
||||
});
|
||||
}
|
||||
|
||||
return { transmissionId, observationId: obsResult.lastInsertRowid };
|
||||
}
|
||||
@@ -270,6 +485,11 @@ function upsertObserver(data) {
|
||||
uptime_secs: data.uptime_secs || null,
|
||||
noise_floor: data.noise_floor || null,
|
||||
});
|
||||
// Update in-memory map for v3
|
||||
if (schemaVersion >= 3 && !observerIdToRowid.has(data.id)) {
|
||||
const row = stmts.getObserverRowid.get(data.id);
|
||||
if (row) observerIdToRowid.set(data.id, row.rowid);
|
||||
}
|
||||
}
|
||||
|
||||
function updateObserverStatus(data) {
|
||||
@@ -592,4 +812,4 @@ function getNodeAnalytics(pubkey, days) {
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = { db, insertTransmission, upsertNode, upsertObserver, updateObserverStatus, getPackets, getPacket, getTransmission, getNodes, getNode, getObservers, getStats, searchNodes, getNodeHealth, getNodeAnalytics };
|
||||
module.exports = { db, schemaVersion, observerIdToRowid, resolveObserverIdx, insertTransmission, upsertNode, upsertObserver, updateObserverStatus, getPackets, getPacket, getTransmission, getNodes, getNode, getObservers, getStats, searchNodes, getNodeHealth, getNodeAnalytics };
|
||||
|
||||
@@ -66,15 +66,28 @@ class PacketStore {
|
||||
|
||||
/** Load from normalized transmissions + observations tables */
|
||||
_loadNormalized() {
|
||||
const rows = this.db.prepare(`
|
||||
SELECT t.id AS transmission_id, t.raw_hex, t.hash, t.first_seen, t.route_type,
|
||||
t.payload_type, t.payload_version, t.decoded_json,
|
||||
o.id AS observation_id, o.observer_id, o.observer_name, o.direction,
|
||||
o.snr, o.rssi, o.score, o.path_json, o.timestamp AS obs_timestamp
|
||||
FROM transmissions t
|
||||
LEFT JOIN observations o ON o.transmission_id = t.id
|
||||
ORDER BY t.first_seen DESC, o.timestamp DESC
|
||||
`).all();
|
||||
// Detect v3 schema (observer_idx instead of observer_id in observations)
|
||||
const obsCols = this.db.pragma('table_info(observations)').map(c => c.name);
|
||||
const isV3 = obsCols.includes('observer_idx');
|
||||
|
||||
const sql = isV3
|
||||
? `SELECT t.id AS transmission_id, t.raw_hex, t.hash, t.first_seen, t.route_type,
|
||||
t.payload_type, t.payload_version, t.decoded_json,
|
||||
o.id AS observation_id, obs.id AS observer_id, obs.name AS observer_name, o.direction,
|
||||
o.snr, o.rssi, o.score, o.path_json, datetime(o.timestamp, 'unixepoch') AS obs_timestamp
|
||||
FROM transmissions t
|
||||
LEFT JOIN observations o ON o.transmission_id = t.id
|
||||
LEFT JOIN observers obs ON obs.rowid = o.observer_idx
|
||||
ORDER BY t.first_seen DESC, o.timestamp DESC`
|
||||
: `SELECT t.id AS transmission_id, t.raw_hex, t.hash, t.first_seen, t.route_type,
|
||||
t.payload_type, t.payload_version, t.decoded_json,
|
||||
o.id AS observation_id, o.observer_id, o.observer_name, o.direction,
|
||||
o.snr, o.rssi, o.score, o.path_json, o.timestamp AS obs_timestamp
|
||||
FROM transmissions t
|
||||
LEFT JOIN observations o ON o.transmission_id = t.id
|
||||
ORDER BY t.first_seen DESC, o.timestamp DESC`;
|
||||
|
||||
const rows = this.db.prepare(sql).all();
|
||||
|
||||
for (const row of rows) {
|
||||
if (this.packets.length >= this.maxPackets && !this.byHash.has(row.hash)) break;
|
||||
|
||||
@@ -19,6 +19,7 @@ node test-regional-filter.js
|
||||
node test-server-helpers.js
|
||||
node test-server-routes.js
|
||||
node test-db.js
|
||||
node test-db-migration.js
|
||||
|
||||
# Integration tests (spin up temp servers)
|
||||
echo ""
|
||||
|
||||
319
test-db-migration.js
Normal file
319
test-db-migration.js
Normal file
@@ -0,0 +1,319 @@
|
||||
'use strict';
|
||||
|
||||
// Test v3 migration: create old-schema DB, run db.js to migrate, verify results
|
||||
const path = require('path');
|
||||
const fs = require('fs');
|
||||
const os = require('os');
|
||||
const { execSync } = require('child_process');
|
||||
const Database = require('better-sqlite3');
|
||||
|
||||
let passed = 0, failed = 0;
|
||||
function assert(cond, msg) {
|
||||
if (cond) { passed++; console.log(` ✅ ${msg}`); }
|
||||
else { failed++; console.error(` ❌ ${msg}`); }
|
||||
}
|
||||
|
||||
console.log('── db.js v3 migration tests ──\n');
|
||||
|
||||
// Helper: create a DB with old (v2) schema and test data
|
||||
function createOldSchemaDB(dbPath) {
|
||||
const db = new Database(dbPath);
|
||||
db.pragma('journal_mode = WAL');
|
||||
db.pragma('foreign_keys = ON');
|
||||
|
||||
db.exec(`
|
||||
CREATE TABLE nodes (
|
||||
public_key TEXT PRIMARY KEY,
|
||||
name TEXT,
|
||||
role TEXT,
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
last_seen TEXT,
|
||||
first_seen TEXT,
|
||||
advert_count INTEGER DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE observers (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT,
|
||||
iata TEXT,
|
||||
last_seen TEXT,
|
||||
first_seen TEXT,
|
||||
packet_count INTEGER DEFAULT 0,
|
||||
model TEXT,
|
||||
firmware TEXT,
|
||||
client_version TEXT,
|
||||
radio TEXT,
|
||||
battery_mv INTEGER,
|
||||
uptime_secs INTEGER,
|
||||
noise_floor INTEGER
|
||||
);
|
||||
|
||||
CREATE TABLE transmissions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
raw_hex TEXT NOT NULL,
|
||||
hash TEXT NOT NULL UNIQUE,
|
||||
first_seen TEXT NOT NULL,
|
||||
route_type INTEGER,
|
||||
payload_type INTEGER,
|
||||
payload_version INTEGER,
|
||||
decoded_json TEXT,
|
||||
created_at TEXT DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
CREATE TABLE observations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
transmission_id INTEGER NOT NULL REFERENCES transmissions(id),
|
||||
hash TEXT NOT NULL,
|
||||
observer_id TEXT,
|
||||
observer_name TEXT,
|
||||
direction TEXT,
|
||||
snr REAL,
|
||||
rssi REAL,
|
||||
score INTEGER,
|
||||
path_json TEXT,
|
||||
timestamp TEXT NOT NULL,
|
||||
created_at TEXT DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
CREATE INDEX idx_transmissions_hash ON transmissions(hash);
|
||||
CREATE INDEX idx_observations_hash ON observations(hash);
|
||||
CREATE INDEX idx_observations_transmission_id ON observations(transmission_id);
|
||||
CREATE INDEX idx_observations_observer_id ON observations(observer_id);
|
||||
CREATE INDEX idx_observations_timestamp ON observations(timestamp);
|
||||
CREATE UNIQUE INDEX idx_observations_dedup ON observations(hash, observer_id, COALESCE(path_json, ''));
|
||||
`);
|
||||
|
||||
// Insert test observers
|
||||
db.prepare(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES (?, ?, ?, ?, ?, ?)`).run(
|
||||
'aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344', 'Observer Alpha', 'SFO',
|
||||
'2025-06-01T12:00:00Z', '2025-01-01T00:00:00Z', 100
|
||||
);
|
||||
db.prepare(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES (?, ?, ?, ?, ?, ?)`).run(
|
||||
'deadbeef12345678deadbeef12345678deadbeef12345678deadbeef12345678', 'Observer Beta', 'LAX',
|
||||
'2025-06-01T11:00:00Z', '2025-02-01T00:00:00Z', 50
|
||||
);
|
||||
|
||||
// Insert test transmissions
|
||||
db.prepare(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES (?, ?, ?, ?, ?, ?)`).run(
|
||||
'0400aabbccdd', 'hash-mig-001', '2025-06-01T10:00:00Z', 1, 4, '{"type":"ADVERT"}'
|
||||
);
|
||||
db.prepare(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES (?, ?, ?, ?, ?, ?)`).run(
|
||||
'0400deadbeef', 'hash-mig-002', '2025-06-01T10:30:00Z', 2, 5, '{"type":"GRP_TXT"}'
|
||||
);
|
||||
|
||||
// Insert test observations (old schema: has hash, observer_id, observer_name, text timestamp)
|
||||
db.prepare(`INSERT INTO observations (transmission_id, hash, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(
|
||||
1, 'hash-mig-001', 'aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344', 'Observer Alpha',
|
||||
'rx', 12.5, -80, 85, '["aabb","ccdd"]', '2025-06-01T10:00:00Z'
|
||||
);
|
||||
db.prepare(`INSERT INTO observations (transmission_id, hash, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(
|
||||
1, 'hash-mig-001', 'deadbeef12345678deadbeef12345678deadbeef12345678deadbeef12345678', 'Observer Beta',
|
||||
'rx', 8.0, -92, 70, '["aabb"]', '2025-06-01T10:01:00Z'
|
||||
);
|
||||
db.prepare(`INSERT INTO observations (transmission_id, hash, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`).run(
|
||||
2, 'hash-mig-002', 'aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344', 'Observer Alpha',
|
||||
'rx', 15.0, -75, 90, null, '2025-06-01T10:30:00Z'
|
||||
);
|
||||
|
||||
db.close();
|
||||
}
|
||||
|
||||
// Helper: require db.js in a child process with a given DB_PATH, return schema info
|
||||
function runDbModule(dbPath) {
|
||||
const scriptPath = path.join(os.tmpdir(), 'meshcore-mig-test-script.js');
|
||||
fs.writeFileSync(scriptPath, `
|
||||
process.env.DB_PATH = ${JSON.stringify(dbPath)};
|
||||
const db = require(${JSON.stringify(path.resolve(__dirname, 'db'))});
|
||||
const cols = db.db.pragma('table_info(observations)').map(c => c.name);
|
||||
const sv = db.db.prepare('SELECT version FROM schema_version ORDER BY version DESC LIMIT 1').get();
|
||||
const obsCount = db.db.prepare('SELECT COUNT(*) as c FROM observations').get().c;
|
||||
const viewRows = db.db.prepare('SELECT * FROM packets_v ORDER BY id').all();
|
||||
const rawObs = db.db.prepare('SELECT * FROM observations ORDER BY id').all();
|
||||
console.log(JSON.stringify({
|
||||
columns: cols,
|
||||
schemaVersion: sv ? sv.version : 0,
|
||||
obsCount,
|
||||
viewRows,
|
||||
rawObs
|
||||
}));
|
||||
db.db.close();
|
||||
`);
|
||||
const result = execSync(`node ${JSON.stringify(scriptPath)}`, {
|
||||
cwd: __dirname,
|
||||
encoding: 'utf8',
|
||||
timeout: 30000,
|
||||
});
|
||||
fs.unlinkSync(scriptPath);
|
||||
const lines = result.trim().split('\n');
|
||||
for (let i = lines.length - 1; i >= 0; i--) {
|
||||
try { return JSON.parse(lines[i]); } catch {}
|
||||
}
|
||||
throw new Error('No JSON output from child process: ' + result);
|
||||
}
|
||||
|
||||
// --- Test 1: Migration from old schema ---
|
||||
console.log('Migration from old schema:');
|
||||
{
|
||||
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'meshcore-mig-test-'));
|
||||
const dbPath = path.join(tmpDir, 'test-mig.db');
|
||||
|
||||
createOldSchemaDB(dbPath);
|
||||
|
||||
// Run db.js which should trigger migration
|
||||
const info = runDbModule(dbPath);
|
||||
|
||||
// Verify schema
|
||||
assert(info.schemaVersion === 3, 'schema version is 3 after migration');
|
||||
assert(info.columns.includes('observer_idx'), 'has observer_idx column');
|
||||
assert(!info.columns.includes('observer_id'), 'no observer_id column');
|
||||
assert(!info.columns.includes('observer_name'), 'no observer_name column');
|
||||
assert(!info.columns.includes('hash'), 'no hash column');
|
||||
|
||||
// Verify row count
|
||||
assert(info.obsCount === 3, `all 3 rows migrated (got ${info.obsCount})`);
|
||||
|
||||
// Verify raw observation data
|
||||
const obs0 = info.rawObs[0];
|
||||
assert(typeof obs0.timestamp === 'number', 'timestamp is integer');
|
||||
assert(obs0.timestamp === Math.floor(new Date('2025-06-01T10:00:00Z').getTime() / 1000), 'timestamp epoch correct');
|
||||
assert(obs0.observer_idx !== null, 'observer_idx populated');
|
||||
|
||||
// Verify view backward compat
|
||||
const vr0 = info.viewRows[0];
|
||||
assert(vr0.observer_id === 'aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344', 'view observer_id correct');
|
||||
assert(vr0.observer_name === 'Observer Alpha', 'view observer_name correct');
|
||||
assert(typeof vr0.timestamp === 'string', 'view timestamp is string');
|
||||
assert(vr0.hash === 'hash-mig-001', 'view hash correct');
|
||||
assert(vr0.snr === 12.5, 'view snr correct');
|
||||
assert(vr0.path_json === '["aabb","ccdd"]', 'view path_json correct');
|
||||
|
||||
// Third row has null path_json
|
||||
const vr2 = info.viewRows[2];
|
||||
assert(vr2.path_json === null, 'null path_json preserved');
|
||||
|
||||
// Verify backup file created
|
||||
assert(fs.existsSync(dbPath + '.pre-v3-backup'), 'backup file exists');
|
||||
|
||||
fs.rmSync(tmpDir, { recursive: true });
|
||||
}
|
||||
|
||||
// --- Test 2: Migration doesn't re-run ---
|
||||
console.log('\nMigration idempotency:');
|
||||
{
|
||||
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'meshcore-mig-test2-'));
|
||||
const dbPath = path.join(tmpDir, 'test-mig2.db');
|
||||
|
||||
createOldSchemaDB(dbPath);
|
||||
|
||||
// First run — triggers migration
|
||||
let info = runDbModule(dbPath);
|
||||
assert(info.schemaVersion === 3, 'first run migrates to v3');
|
||||
|
||||
// Second run — should NOT re-run migration (no backup overwrite, same data)
|
||||
const backupMtime = fs.statSync(dbPath + '.pre-v3-backup').mtimeMs;
|
||||
info = runDbModule(dbPath);
|
||||
assert(info.schemaVersion === 3, 'second run still v3');
|
||||
assert(info.obsCount === 3, 'rows still intact');
|
||||
|
||||
fs.rmSync(tmpDir, { recursive: true });
|
||||
}
|
||||
|
||||
// --- Test 3: Backup failure aborts migration ---
|
||||
console.log('\nBackup failure aborts migration:');
|
||||
{
|
||||
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'meshcore-mig-test3-'));
|
||||
const dbPath = path.join(tmpDir, 'test-mig3.db');
|
||||
|
||||
createOldSchemaDB(dbPath);
|
||||
|
||||
// Create backup path as a directory so copyFileSync fails
|
||||
fs.mkdirSync(dbPath + '.pre-v3-backup');
|
||||
|
||||
// Run db.js — migration should abort, old schema preserved
|
||||
const info = runDbModule(dbPath);
|
||||
|
||||
// Old schema should be preserved
|
||||
assert(info.columns.includes('observer_id'), 'old observer_id column preserved');
|
||||
assert(info.schemaVersion < 3, 'schema version not updated');
|
||||
assert(info.obsCount === 3, 'old rows still intact');
|
||||
|
||||
fs.rmSync(tmpDir, { recursive: true });
|
||||
}
|
||||
|
||||
// --- Test 4: v3 ingestion via child process ---
|
||||
console.log('\nv3 ingestion test:');
|
||||
{
|
||||
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'meshcore-mig-test4-'));
|
||||
const dbPath = path.join(tmpDir, 'test-v3-ingest.db');
|
||||
|
||||
const scriptPath = path.join(os.tmpdir(), 'meshcore-ingest-test-script.js');
|
||||
fs.writeFileSync(scriptPath, `
|
||||
process.env.DB_PATH = ${JSON.stringify(dbPath)};
|
||||
const db = require(${JSON.stringify(path.resolve(__dirname, 'db'))});
|
||||
|
||||
db.upsertObserver({ id: 'test-obs', name: 'Test Obs' });
|
||||
|
||||
const r = db.insertTransmission({
|
||||
raw_hex: '0400ff',
|
||||
hash: 'h-001',
|
||||
timestamp: '2025-06-01T12:00:00Z',
|
||||
observer_id: 'test-obs',
|
||||
observer_name: 'Test Obs',
|
||||
direction: 'rx',
|
||||
snr: 10,
|
||||
rssi: -85,
|
||||
path_json: '["aa"]',
|
||||
route_type: 1,
|
||||
payload_type: 4,
|
||||
});
|
||||
|
||||
const r2 = db.insertTransmission({
|
||||
raw_hex: '0400ff',
|
||||
hash: 'h-001',
|
||||
timestamp: '2025-06-01T12:00:00Z',
|
||||
observer_id: 'test-obs',
|
||||
direction: 'rx',
|
||||
snr: 10,
|
||||
rssi: -85,
|
||||
path_json: '["aa"]',
|
||||
});
|
||||
|
||||
const pkt = db.db.prepare('SELECT * FROM packets_v WHERE hash = ?').get('h-001');
|
||||
|
||||
console.log(JSON.stringify({
|
||||
r1_ok: r !== null && r.transmissionId > 0,
|
||||
r2_deduped: r2.observationId === 0,
|
||||
obs_count: db.db.prepare('SELECT COUNT(*) as c FROM observations').get().c,
|
||||
view_observer_id: pkt.observer_id,
|
||||
view_observer_name: pkt.observer_name,
|
||||
view_ts_type: typeof pkt.timestamp,
|
||||
}));
|
||||
db.db.close();
|
||||
`);
|
||||
|
||||
const result = execSync(`node ${JSON.stringify(scriptPath)}`, {
|
||||
cwd: __dirname, encoding: 'utf8', timeout: 30000,
|
||||
});
|
||||
fs.unlinkSync(scriptPath);
|
||||
const lines = result.trim().split('\n');
|
||||
let info;
|
||||
for (let i = lines.length - 1; i >= 0; i--) {
|
||||
try { info = JSON.parse(lines[i]); break; } catch {}
|
||||
}
|
||||
|
||||
assert(info.r1_ok, 'first insertion succeeded');
|
||||
assert(info.r2_deduped, 'duplicate caught by dedup');
|
||||
assert(info.obs_count === 1, 'only one observation row');
|
||||
assert(info.view_observer_id === 'test-obs', 'view resolves observer_id');
|
||||
assert(info.view_observer_name === 'Test Obs', 'view resolves observer_name');
|
||||
assert(info.view_ts_type === 'string', 'view timestamp is string');
|
||||
|
||||
fs.rmSync(tmpDir, { recursive: true });
|
||||
}
|
||||
|
||||
console.log(`\n═══════════════════════════════════════`);
|
||||
console.log(` PASSED: ${passed}`);
|
||||
console.log(` FAILED: ${failed}`);
|
||||
console.log(`═══════════════════════════════════════`);
|
||||
if (failed > 0) process.exit(1);
|
||||
110
test-db.js
110
test-db.js
@@ -252,9 +252,113 @@ console.log('\ngetNodeAnalytics:');
|
||||
// --- seed ---
|
||||
console.log('\nseed:');
|
||||
{
|
||||
// Already has data, should return false
|
||||
const result = db.seed();
|
||||
assert(result === false, 'seed returns false when data exists');
|
||||
if (typeof db.seed === 'function') {
|
||||
// Already has data, should return false
|
||||
const result = db.seed();
|
||||
assert(result === false, 'seed returns false when data exists');
|
||||
} else {
|
||||
console.log(' (skipped — seed not exported)');
|
||||
}
|
||||
}
|
||||
|
||||
// --- v3 schema tests (fresh DB should be v3) ---
|
||||
console.log('\nv3 schema:');
|
||||
{
|
||||
assert(db.schemaVersion >= 3, 'fresh DB creates v3 schema');
|
||||
|
||||
// observations table should have observer_idx, not observer_id
|
||||
const cols = db.db.pragma('table_info(observations)').map(c => c.name);
|
||||
assert(cols.includes('observer_idx'), 'observations has observer_idx column');
|
||||
assert(!cols.includes('observer_id'), 'observations does NOT have observer_id column');
|
||||
assert(!cols.includes('observer_name'), 'observations does NOT have observer_name column');
|
||||
assert(!cols.includes('hash'), 'observations does NOT have hash column');
|
||||
assert(!cols.includes('created_at'), 'observations does NOT have created_at column');
|
||||
|
||||
// timestamp should be integer
|
||||
const obsRow = db.db.prepare('SELECT typeof(timestamp) as t FROM observations LIMIT 1').get();
|
||||
if (obsRow) {
|
||||
assert(obsRow.t === 'integer', 'timestamp is stored as integer');
|
||||
}
|
||||
|
||||
// packets_v view should still expose observer_id, observer_name, ISO timestamp
|
||||
const viewRow = db.db.prepare('SELECT * FROM packets_v LIMIT 1').get();
|
||||
if (viewRow) {
|
||||
assert('observer_id' in viewRow, 'packets_v exposes observer_id');
|
||||
assert('observer_name' in viewRow, 'packets_v exposes observer_name');
|
||||
assert(typeof viewRow.timestamp === 'string', 'packets_v timestamp is ISO string');
|
||||
}
|
||||
|
||||
// schema_version table exists with version 3
|
||||
const sv = db.db.prepare('SELECT version FROM schema_version ORDER BY version DESC LIMIT 1').get();
|
||||
assert(sv && sv.version === 3, 'schema_version table has version 3');
|
||||
}
|
||||
|
||||
// --- v3 ingestion: observer resolved via observer_idx ---
|
||||
console.log('\nv3 ingestion with observer resolution:');
|
||||
{
|
||||
// Insert a new observer
|
||||
db.upsertObserver({ id: 'obs-v3-test', name: 'V3 Test Observer' });
|
||||
|
||||
// Insert observation referencing that observer
|
||||
const result = db.insertTransmission({
|
||||
raw_hex: '0400deadbeef',
|
||||
hash: 'hash-v3-001',
|
||||
timestamp: '2025-06-01T12:00:00Z',
|
||||
observer_id: 'obs-v3-test',
|
||||
observer_name: 'V3 Test Observer',
|
||||
direction: 'rx',
|
||||
snr: 12.0,
|
||||
rssi: -80,
|
||||
route_type: 1,
|
||||
payload_type: 4,
|
||||
path_json: '["aabb"]',
|
||||
});
|
||||
assert(result !== null, 'v3 insertion succeeded');
|
||||
assert(result.transmissionId > 0, 'v3 has transmissionId');
|
||||
|
||||
// Verify via packets_v view
|
||||
const pkt = db.db.prepare('SELECT * FROM packets_v WHERE hash = ?').get('hash-v3-001');
|
||||
assert(pkt !== null, 'v3 packet found via view');
|
||||
assert(pkt.observer_id === 'obs-v3-test', 'v3 observer_id resolved in view');
|
||||
assert(pkt.observer_name === 'V3 Test Observer', 'v3 observer_name resolved in view');
|
||||
assert(typeof pkt.timestamp === 'string', 'v3 timestamp is ISO string in view');
|
||||
assert(pkt.timestamp.includes('2025-06-01'), 'v3 timestamp date correct');
|
||||
|
||||
// Raw observation should have integer timestamp
|
||||
const obs = db.db.prepare('SELECT * FROM observations ORDER BY id DESC LIMIT 1').get();
|
||||
assert(typeof obs.timestamp === 'number', 'v3 raw observation timestamp is integer');
|
||||
assert(obs.observer_idx !== null, 'v3 observation has observer_idx');
|
||||
}
|
||||
|
||||
// --- v3 dedup ---
|
||||
console.log('\nv3 dedup:');
|
||||
{
|
||||
// Insert same observation again — should be deduped
|
||||
const result = db.insertTransmission({
|
||||
raw_hex: '0400deadbeef',
|
||||
hash: 'hash-v3-001',
|
||||
timestamp: '2025-06-01T12:00:00Z',
|
||||
observer_id: 'obs-v3-test',
|
||||
direction: 'rx',
|
||||
snr: 12.0,
|
||||
rssi: -80,
|
||||
path_json: '["aabb"]',
|
||||
});
|
||||
assert(result.observationId === 0, 'duplicate caught by in-memory dedup');
|
||||
|
||||
// Different observer = not a dupe
|
||||
db.upsertObserver({ id: 'obs-v3-test-2', name: 'V3 Test Observer 2' });
|
||||
const result2 = db.insertTransmission({
|
||||
raw_hex: '0400deadbeef',
|
||||
hash: 'hash-v3-001',
|
||||
timestamp: '2025-06-01T12:01:00Z',
|
||||
observer_id: 'obs-v3-test-2',
|
||||
direction: 'rx',
|
||||
snr: 9.0,
|
||||
rssi: -88,
|
||||
path_json: '["ccdd"]',
|
||||
});
|
||||
assert(result2.observationId > 0, 'different observer is not a dupe');
|
||||
}
|
||||
|
||||
cleanup();
|
||||
|
||||
@@ -15,6 +15,10 @@ function createMockDb() {
|
||||
let obsIdCounter = 1000;
|
||||
return {
|
||||
db: {
|
||||
pragma: (query) => {
|
||||
if (query.includes('table_info(observations)')) return [{ name: 'observer_idx' }];
|
||||
return [];
|
||||
},
|
||||
prepare: (sql) => ({
|
||||
get: (...args) => {
|
||||
if (sql.includes('sqlite_master')) return { name: 'transmissions' };
|
||||
|
||||
Reference in New Issue
Block a user