From eea095d97d4cc53e644da82aa91f5b41917d3774 Mon Sep 17 00:00:00 2001 From: levibliz Date: Tue, 18 Aug 2026 01:09:29 +0100 Subject: [PATCH] feat(storage): implement time-series compaction with downsampling and retention Add automated downsampling, retention policies, and continuous aggregates for cost-effective long-term historical storage of location events. - Extend StorageAdapter interface with compact() and getCompactionStatus() - Add resolution parameter to queryRoom() for tiered data access (raw, 1m, 1h, 1d, auto) - Implement declarative partitioning by day on location_events - Add location_events_1m, _1h, _1d downsample/aggregate tables - Implement compaction pipeline: partition drop, 1m downsampling, 1h aggregates, 1d rollups, tier retention trimming - Add transparent query routing with auto-resolution selection - Implement MemoryAdapter compaction for testing without PostgreSQL - Add migration-compaction.sql for existing deployments - Update docker-compose.yml for TimescaleDB support - Add compaction configuration env vars - Add comprehensive test suite (storage-compaction.test.js) Closes #257 Closes #254 Closes #252 Closes #251 --- .env.example | 25 + docker-compose.yml | 13 +- src/index.js | 3 +- src/rate-limiter.js | 2 + src/server.js | 53 +- src/storage/adapter.js | 60 +- src/storage/index.js | 74 +- src/storage/memory.js | 485 +++++++++++- src/storage/migration-compaction.sql | 279 +++++++ src/storage/postgres.js | 1018 ++++++++++++++++++++++++-- tests/storage-compaction.test.js | 617 ++++++++++++++++ 11 files changed, 2506 insertions(+), 123 deletions(-) create mode 100644 src/storage/migration-compaction.sql create mode 100644 tests/storage-compaction.test.js diff --git a/.env.example b/.env.example index 483fbbe..379dc43 100644 --- a/.env.example +++ b/.env.example @@ -14,3 +14,28 @@ MAX_MESSAGES_PER_SECOND=100 # Maximum new WebSocket connections allowed per IP address per minute CONN_RATE_LIMIT=30 + +# Storage adapter: "postgres" | "memory" | "none" (default: "memory") +STORAGE_ADAPTER=memory + +# PostgreSQL connection string (required when STORAGE_ADAPTER=postgres) +# DATABASE_URL=postgresql://tracker:devpassword@localhost:5432/spatial_tracking + +# Storage tuning +STORAGE_POOL_SIZE=10 +STORAGE_BATCH_SIZE=100 +STORAGE_FLUSH_INTERVAL_MS=1000 +STORAGE_MAX_BUFFER_SIZE=10000 + +# Time-series compaction retention policies (days) +STORAGE_RAW_RETENTION_DAYS=7 +STORAGE_1M_RETENTION_DAYS=90 +STORAGE_1H_RETENTION_DAYS=365 +STORAGE_1D_RETENTION_DAYS=2555 + +# Compaction job settings +STORAGE_COMPACTION_INTERVAL_MS=300000 +STORAGE_COMPACTION_BATCH_SIZE=10000 + +# Enable TimescaleDB features (set to "true" when using timescale/timescaledb image) +TIMESCALEDB_ENABLED=false diff --git a/docker-compose.yml b/docker-compose.yml index 01cd580..9fc86a2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -10,10 +10,21 @@ services: WS_HEARTBEAT_MS: "30000" MAX_PAYLOAD_BYTES: "1024" LOG_LEVEL: "info" + STORAGE_ADAPTER: "postgres" + DATABASE_URL: "postgresql://tracker:devpassword@postgres:5432/spatial_tracking" + STORAGE_RAW_RETENTION_DAYS: "7" + STORAGE_1M_RETENTION_DAYS: "90" + STORAGE_1H_RETENTION_DAYS: "365" + STORAGE_1D_RETENTION_DAYS: "2555" + STORAGE_COMPACTION_INTERVAL_MS: "300000" + STORAGE_COMPACTION_BATCH_SIZE: "10000" + TIMESCALEDB_ENABLED: "false" + depends_on: + - postgres restart: unless-stopped postgres: - image: postgres:16-alpine + image: timescale/timescaledb:latest-pg16 environment: POSTGRES_DB: spatial_tracking POSTGRES_USER: tracker diff --git a/src/index.js b/src/index.js index b03bcf9..fd5b79c 100644 --- a/src/index.js +++ b/src/index.js @@ -34,9 +34,8 @@ if (isNaN(config.maxPayloadBytes) || config.maxPayloadBytes < 1) { let wss; let httpServer; let markShuttingDown; -let sessionManager; try { - ({ wss, httpServer, markShuttingDown, sessionManager } = createServer(config)); + ({ wss, httpServer, markShuttingDown } = createServer(config)); } catch (err) { logger.error("Failed to start server", { error: err.message }); process.exit(1); diff --git a/src/rate-limiter.js b/src/rate-limiter.js index fe5b805..8c4f182 100644 --- a/src/rate-limiter.js +++ b/src/rate-limiter.js @@ -83,3 +83,5 @@ export function createRateLimiter(maxPerSecond) { get size() { return windows.size; }, + }; +} diff --git a/src/server.js b/src/server.js index 4285fd6..516c712 100644 --- a/src/server.js +++ b/src/server.js @@ -8,9 +8,7 @@ import { logger } from "./logger.js"; import { createRateLimiter } from "./rate-limiter.js"; import { createConnRateLimiter } from "./conn-rate-limiter.js"; import { VALIDATION_ERROR } from "./errors.js"; - -export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) { -import { createRateLimiter } from "./rate-limiter.js"; +import { SessionManager } from "./session-manager.js"; export function createServer({ port, @@ -23,18 +21,35 @@ export function createServer({ maxBufferBytes: _maxBufferBytes, maxDedupEntries: _maxDedupEntries, } = {}) { + let isShuttingDown = false; + const markShuttingDown = () => { isShuttingDown = true; }; + + const metrics = { + messages: { location_update: 0, join_room: 0, leave_room: 0 }, + rateLimitRejections: { connection: 0 }, + authFailures: 0, + sessionResumption: { success: 0, decrypt_failed: 0, expired: 0, mismatch: 0, new_session: 0 }, + eventLoopLagMs: 0, + }; + + const sessionManager = new SessionManager(); + + function safeSend(ws, data) { + if (ws.readyState === 1) { + ws.send(typeof data === "string" ? data : JSON.stringify(data)); + } + } + const server = http.createServer((req, res) => { - let url; + let pathname; try { - url = new URL(req.url, `http://${req.headers.host || "localhost"}`); + pathname = new URL(req.url, `http://${req.headers.host || "localhost"}`).pathname; } catch { res.writeHead(400, { "Content-Type": "application/json" }); res.end(JSON.stringify({ error: "Bad Request" })); return; } - const pathname = new URL(req.url, `http://${req.headers.host ?? "localhost"}`).pathname; - if (pathname === "/health" || pathname === "/healthz") { if (isShuttingDown && pathname === "/healthz") { res.writeHead(503, { "Content-Type": "application/json" }); @@ -93,15 +108,13 @@ export function createServer({ } }); - httpServer.listen(port ?? 8080); + server.listen(port ?? 8080); const wss = new WebSocketServer({ - server: httpServer, + server, maxPayload: maxPayloadBytes ?? 1024, }); - server.listen(port ?? 8080); - const rooms = new RoomManager(); const rateLimiter = createRateLimiter(); const connRateLimiter = createConnRateLimiter(connRateLimit); @@ -116,13 +129,12 @@ export function createServer({ ws.send(JSON.stringify({ type: "error", payload: { message, code } })); } - wss.on("connection", (ws, req) => { + wss.on("connection", async (ws, req) => { const clientId = uuid(); ws.isAlive = true; const ip = req.socket.remoteAddress; - // Per-IP connection rate limit (new connections per minute) if (!connRateLimiter.check(ip)) { logger.warn("Connection rate limit exceeded", { ip }); metrics.rateLimitRejections.connection++; @@ -159,13 +171,13 @@ export function createServer({ return; } - const actualClientId = authResult.clientId ?? clientId; + let actualClientId = authResult.clientId ?? clientId; ws._clientId = actualClientId; logger.info("Client connected", { clientId: actualClientId, ip }); ws.on("pong", heartbeat); - ws.on("message", (raw) => { + ws.on("message", async (raw) => { if (!rateLimiter.check(actualClientId)) { logger.warn("Message rate limit exceeded", { clientId: actualClientId }); ws.send(JSON.stringify({ type: "error", payload: { message: "Rate limit exceeded" } })); @@ -185,6 +197,7 @@ export function createServer({ switch (msg.type) { case "join_room": { + metrics.messages.join_room++; const joinResult = rooms.join(actualClientId, msg.roomId, ws); if (!joinResult.ok && joinResult.reason === 'ROOM_FULL') { logger.warn("Room is full", { clientId: actualClientId, roomId: msg.roomId }); @@ -212,8 +225,8 @@ export function createServer({ break; } case "leave_room": { - rooms.leave(actualClientId, msg.roomId); metrics.messages.leave_room++; + rooms.leave(actualClientId, msg.roomId); logger.info("Client left room", { clientId: actualClientId, roomId: msg.roomId }); safeSend(ws, { type: "room_left", payload: { roomId: msg.roomId } }); @@ -268,7 +281,7 @@ export function createServer({ } }); - ws.on("close", (code, reason) => { + ws.on("close", () => { const currentRooms = rooms.getClientRooms(actualClientId); const roomStates = Array.from(currentRooms).map((roomId) => ({ roomId, @@ -299,8 +312,6 @@ export function createServer({ } logger.info("Client disconnected", { clientId: actualClientId, - code, - reason: reason?.toString() ?? "unknown", }); }); @@ -324,8 +335,8 @@ export function createServer({ wss.on("close", () => { clearInterval(heartbeatInterval); - httpServer.close(); + server.close(); }); - return { wss, server, rooms, ipConnectionCount, rateLimiter }; + return { wss, httpServer: server, rooms, ipConnectionCount, rateLimiter, markShuttingDown }; } diff --git a/src/storage/adapter.js b/src/storage/adapter.js index ece21a2..57c24d6 100644 --- a/src/storage/adapter.js +++ b/src/storage/adapter.js @@ -36,16 +36,54 @@ * Options for time-range / limit queries. * * @typedef {object} QueryOptions - * @property {Date} [from] - Start of the time range (inclusive). - * @property {Date} [to] - End of the time range (inclusive). - * @property {number} [limit] - Maximum number of events to return. + * @property {Date} [from] - Start of the time range (inclusive). + * @property {Date} [to] - End of the time range (inclusive). + * @property {number} [limit] - Maximum number of events to return. + * @property {string} [resolution] - Data resolution: "raw" | "1m" | "1h" | "1d" | "auto" (default: "auto"). + */ + +/** + * Resolution tier identifier. + * @typedef {"raw"|"1m"|"1h"|"1d"|"auto"} Resolution + */ + +/** + * Result of a compaction run. + * + * @typedef {object} CompactionResult + * @property {number} rawDeleted - Number of raw rows deleted (or partitions dropped). + * @property {number} downsample1m - Number of 1-minute downsample rows written. + * @property {number} aggregate1h - Number of 1-hour aggregate rows written. + * @property {number} aggregate1d - Number of 1-day aggregate rows written. + * @property {number} durationMs - Total compaction duration in milliseconds. + * @property {string} [error] - Error message if compaction partially failed. + */ + +/** + * Status of a single retention tier. + * + * @typedef {object} TierStatus + * @property {string} name - Tier name: "raw", "1m", "1h", "1d". + * @property {number} rows - Approximate row count. + * @property {number} sizeBytes- Approximate storage size in bytes. + * @property {string|null} oldest- ISO 8601 timestamp of the oldest row, or null. + * @property {string|null} newest- ISO 8601 timestamp of the newest row, or null. + */ + +/** + * Compaction subsystem status. + * + * @typedef {object} CompactionStatus + * @property {string|null} lastRun - ISO 8601 timestamp of the last completed compaction run. + * @property {string|null} nextRun - ISO 8601 timestamp of the next scheduled run. + * @property {TierStatus[]} tiers - Per-tier statistics. */ /** * The StorageAdapter interface. * * Every concrete adapter (PostgresAdapter, MemoryAdapter, …) must implement - * all five methods below. Methods are async — callers must await them or + * all methods below. Methods are async — callers must await them or * handle the returned Promise. * * @typedef {object} StorageAdapter @@ -57,7 +95,8 @@ * * @property {function(string, QueryOptions=): Promise} queryRoom * Retrieve historical location events for a given room, optionally filtered - * by time range and capped to a maximum result count. + * by time range, capped to a maximum result count, and resolved to a specific + * data tier via the resolution parameter. * * @property {function(SpatialBounds, {limit?: number}=): Promise} querySpatial * Return events whose coordinates fall within the supplied bounding box. @@ -71,6 +110,15 @@ * @property {function(): Promise} close * Release all resources held by the adapter (connections, timers, …). * Must be idempotent — calling it multiple times must not throw. + * + * @property {function(object=): Promise} compact + * Run the compaction pipeline: drop expired raw partitions, compute + * 1-minute downsamples, compute 1-hour aggregates, compute 1-day rollups. + * Accepts optional retention overrides. Returns a summary of work done. + * + * @property {function(): Promise} getCompactionStatus + * Return current compaction status including last/next run times and + * per-tier statistics (row count, size, time range). */ /** @@ -82,7 +130,7 @@ * @throws {TypeError} When one or more required methods are absent. */ export function assertStorageAdapter(adapter) { - const required = ["writeBatch", "queryRoom", "querySpatial", "getLatest", "close"]; + const required = ["writeBatch", "queryRoom", "querySpatial", "getLatest", "close", "compact", "getCompactionStatus"]; const missing = required.filter((m) => typeof adapter[m] !== "function"); if (missing.length > 0) { throw new TypeError( diff --git a/src/storage/index.js b/src/storage/index.js index c25c06a..da212ba 100644 --- a/src/storage/index.js +++ b/src/storage/index.js @@ -10,12 +10,19 @@ * "none" — A no-op adapter that silently discards all writes. * * Environment variables: - * STORAGE_ADAPTER "postgres" | "memory" | "none" (default: "memory") - * DATABASE_URL Postgres connection string (required for "postgres") - * STORAGE_BATCH_SIZE Number of events per bulk insert (default: 100) - * STORAGE_FLUSH_INTERVAL_MS Ms between periodic flushes (default: 1000) - * STORAGE_MAX_BUFFER_SIZE Hard cap on the write buffer (default: 10000) - * STORAGE_POOL_SIZE Max Postgres pool connections (default: 10) + * STORAGE_ADAPTER "postgres" | "memory" | "none" (default: "memory") + * DATABASE_URL Postgres connection string (required for "postgres") + * STORAGE_POOL_SIZE Max Postgres pool connections (default: 10) + * STORAGE_BATCH_SIZE Number of events per bulk insert (default: 100) + * STORAGE_FLUSH_INTERVAL_MS Ms between periodic flushes (default: 1000) + * STORAGE_MAX_BUFFER_SIZE Hard cap on the write buffer (default: 10000) + * STORAGE_RAW_RETENTION_DAYS Days to keep raw events (default: 7) + * STORAGE_1M_RETENTION_DAYS Days to keep 1m downsample (default: 90) + * STORAGE_1H_RETENTION_DAYS Days to keep 1h aggregates (default: 365) + * STORAGE_1D_RETENTION_DAYS Days to keep 1d rollups (default: 2555) + * STORAGE_COMPACTION_INTERVAL_MS Compaction interval in ms (default: 300000) + * STORAGE_COMPACTION_BATCH_SIZE Batch size for deletion loops (default: 10000) + * TIMESCALEDB_ENABLED Enable TimescaleDB features (default: false) */ import { MemoryAdapter } from "./memory.js"; @@ -35,6 +42,21 @@ class NoneAdapter { async querySpatial(_bounds, _options) { return []; } async getLatest(_roomId) { return null; } async close() {} + async compact(_overrides) { + return { rawDeleted: 0, downsample1m: 0, aggregate1h: 0, aggregate1d: 0, durationMs: 0 }; + } + async getCompactionStatus() { + return { + lastRun: null, + nextRun: null, + tiers: [ + { name: "raw", rows: 0, sizeBytes: 0, oldest: null, newest: null }, + { name: "1m", rows: 0, sizeBytes: 0, oldest: null, newest: null }, + { name: "1h", rows: 0, sizeBytes: 0, oldest: null, newest: null }, + { name: "1d", rows: 0, sizeBytes: 0, oldest: null, newest: null }, + ], + }; + } } /** @@ -42,12 +64,19 @@ class NoneAdapter { * (or environment variables when no config object is provided). * * @param {object} [config={}] - * @param {string} [config.adapter] - Override STORAGE_ADAPTER env var. - * @param {string} [config.connectionString] - Override DATABASE_URL env var. - * @param {number} [config.poolSize] - Override STORAGE_POOL_SIZE env var. - * @param {number} [config.batchSize] - Override STORAGE_BATCH_SIZE env var. - * @param {number} [config.flushIntervalMs] - Override STORAGE_FLUSH_INTERVAL_MS env var. - * @param {number} [config.maxBufferSize] - Override STORAGE_MAX_BUFFER_SIZE env var. + * @param {string} [config.adapter] - Override STORAGE_ADAPTER env var. + * @param {string} [config.connectionString] - Override DATABASE_URL env var. + * @param {number} [config.poolSize] - Override STORAGE_POOL_SIZE env var. + * @param {number} [config.batchSize] - Override STORAGE_BATCH_SIZE env var. + * @param {number} [config.flushIntervalMs] - Override STORAGE_FLUSH_INTERVAL_MS env var. + * @param {number} [config.maxBufferSize] - Override STORAGE_MAX_BUFFER_SIZE env var. + * @param {number} [config.rawRetentionDays] - Override STORAGE_RAW_RETENTION_DAYS env var. + * @param {number} [config.downsample1mRetentionDays] - Override STORAGE_1M_RETENTION_DAYS env var. + * @param {number} [config.downsample1hRetentionDays] - Override STORAGE_1H_RETENTION_DAYS env var. + * @param {number} [config.aggregate1dRetentionDays] - Override STORAGE_1D_RETENTION_DAYS env var. + * @param {number} [config.compactionIntervalMs] - Override STORAGE_COMPACTION_INTERVAL_MS env var. + * @param {number} [config.compactionBatchSize] - Override STORAGE_COMPACTION_BATCH_SIZE env var. + * @param {boolean} [config.timescaleDbEnabled] - Override TIMESCALEDB_ENABLED env var. * @returns {import("./adapter.js").StorageAdapter} * @throws {Error} When an unrecognised adapter name is supplied. */ @@ -85,6 +114,27 @@ export function createStorageAdapter(config = {}) { maxBufferSize: config.maxBufferSize ?? parseInt(process.env.STORAGE_MAX_BUFFER_SIZE ?? "10000", 10), + rawRetentionDays: + config.rawRetentionDays ?? + parseInt(process.env.STORAGE_RAW_RETENTION_DAYS ?? "7", 10), + downsample1mRetentionDays: + config.downsample1mRetentionDays ?? + parseInt(process.env.STORAGE_1M_RETENTION_DAYS ?? "90", 10), + downsample1hRetentionDays: + config.downsample1hRetentionDays ?? + parseInt(process.env.STORAGE_1H_RETENTION_DAYS ?? "365", 10), + aggregate1dRetentionDays: + config.aggregate1dRetentionDays ?? + parseInt(process.env.STORAGE_1D_RETENTION_DAYS ?? "2555", 10), + compactionIntervalMs: + config.compactionIntervalMs ?? + parseInt(process.env.STORAGE_COMPACTION_INTERVAL_MS ?? "300000", 10), + compactionBatchSize: + config.compactionBatchSize ?? + parseInt(process.env.STORAGE_COMPACTION_BATCH_SIZE ?? "10000", 10), + timescaleDbEnabled: + config.timescaleDbEnabled ?? + (process.env.TIMESCALEDB_ENABLED === "true"), }); break; } diff --git a/src/storage/memory.js b/src/storage/memory.js index 93d89f3..a113637 100644 --- a/src/storage/memory.js +++ b/src/storage/memory.js @@ -5,10 +5,74 @@ * plain JavaScript array and is lost when the process exits. Every method * of the StorageAdapter interface is implemented so test suites can run the * same contract tests against MemoryAdapter and PostgresAdapter. + * + * Includes in-memory implementations of compaction, downsample, and + * resolution-aware queryRoom for testing without PostgreSQL. */ import { v4 as uuid } from "uuid"; +/** + * Compute median of a numeric array. + * @param {number[]} arr + * @returns {number} + */ +function median(arr) { + if (arr.length === 0) return 0; + const sorted = [...arr].sort((a, b) => a - b); + const mid = Math.floor(sorted.length / 2); + return sorted.length % 2 !== 0 ? sorted[mid] : (sorted[mid - 1] + sorted[mid]) / 2; +} + +/** + * Compute haversine distance between two lat/lon points in metres. + * @param {number} lat1 + * @param {number} lon1 + * @param {number} lat2 + * @param {number} lon2 + * @returns {number} + */ +function haversine(lat1, lon1, lat2, lon2) { + const R = 6371000; + const toRad = (d) => (d * Math.PI) / 180; + const dLat = toRad(lat2 - lat1); + const dLon = toRad(lon2 - lon1); + const a = + Math.sin(dLat / 2) ** 2 + + Math.cos(toRad(lat1)) * Math.cos(toRad(lat2)) * Math.sin(dLon / 2) ** 2; + return R * 2 * Math.asin(Math.sqrt(a)); +} + +/** + * Truncate a Date to the start of its minute. + * @param {Date|string} d + * @returns {number} Unix timestamp (ms) at start of minute. + */ +function truncateToMinute(d) { + const t = typeof d === "string" ? new Date(d).getTime() : d.getTime(); + return Math.floor(t / 60000) * 60000; +} + +/** + * Truncate a Date to the start of its hour. + * @param {Date|string} d + * @returns {number} Unix timestamp (ms) at start of hour. + */ +function truncateToHour(d) { + const t = typeof d === "string" ? new Date(d).getTime() : d.getTime(); + return Math.floor(t / 3600000) * 3600000; +} + +/** + * Truncate a Date to the start of its day (UTC). + * @param {Date|string} d + * @returns {number} Unix timestamp (ms) at start of day. + */ +function truncateToDay(d) { + const t = typeof d === "string" ? new Date(d).getTime() : d.getTime(); + return Math.floor(t / 86400000) * 86400000; +} + /** * @implements {import("./adapter.js").StorageAdapter} */ @@ -17,8 +81,31 @@ export class MemoryAdapter { /** @type {import("./adapter.js").LocationEvent[]} */ this._events = []; + /** @type {object[]} In-memory 1-minute downsample rows. */ + this._downsample1m = []; + + /** @type {object[]} In-memory 1-hour aggregate rows. */ + this._aggregate1h = []; + + /** @type {object[]} In-memory 1-day aggregate rows. */ + this._aggregate1d = []; + /** Whether close() has been called. */ this._closed = false; + + /** Retention configuration. */ + this._retention = { + rawRetentionDays: 7, + downsample1mRetentionDays: 90, + downsample1hRetentionDays: 365, + aggregate1dRetentionDays: 2555, + }; + + /** Last compaction run timestamp. */ + this._lastCompactionRun = null; + + /** Next scheduled compaction run timestamp. */ + this._nextCompactionRun = null; } /** @@ -45,6 +132,7 @@ export class MemoryAdapter { /** * Returns events for a specific room, optionally filtered by time range, * ordered by `timestamp` ascending and capped to `limit`. + * Supports resolution parameter for tiered data access. * * @param {string} roomId * @param {import("./adapter.js").QueryOptions} [options={}] @@ -53,8 +141,67 @@ export class MemoryAdapter { async queryRoom(roomId, options = {}) { if (this._closed) throw new Error("MemoryAdapter is closed"); - const { from, to, limit } = options; + const { from, to, limit, resolution = "auto" } = options; + + let resolved = resolution; + if (resolution === "auto") { + resolved = this._autoResolve(from, to, limit); + } + + if (resolved === "raw") { + return this._queryRoomRaw(roomId, from, to, limit); + } + if (resolved === "1m") { + return this._queryRoomDownsample(roomId, from, to, limit, "1m"); + } + if (resolved === "1h") { + return this._queryRoomAggregate(roomId, from, to, limit, "1h"); + } + if (resolved === "1d") { + return this._queryRoomAggregate(roomId, from, to, limit, "1d"); + } + + return this._queryRoomRaw(roomId, from, to, limit); + } + + /** + * Auto-select resolution based on time range and desired point count. + * @private + * @param {Date} [from] + * @param {Date} [to] + * @param {number} [limit] + * @returns {string} + */ + _autoResolve(from, to, limit) { + // When no time range is specified or only one bound is given, + // default to raw for backwards compatibility + if (!from || !to) return "raw"; + + const maxPoints = limit || 1000; + const fromMs = from.getTime(); + const toMs = to.getTime(); + const rangeMs = toMs - fromMs; + + // Estimate points at each resolution + if (rangeMs <= 0) return "raw"; + + const rawPoints = Math.floor(rangeMs / 1000); // 1 point/sec estimate + if (rawPoints <= maxPoints) return "raw"; + + const m1Points = Math.floor(rangeMs / 60000); + if (m1Points <= maxPoints) return "1m"; + + const h1Points = Math.floor(rangeMs / 3600000); + if (h1Points <= maxPoints) return "1h"; + + return "1d"; + } + /** + * Query raw events for a room. + * @private + */ + _queryRoomRaw(roomId, from, to, limit) { let results = this._events.filter((e) => { if (e.roomId !== roomId) return false; const ts = new Date(e.timestamp).getTime(); @@ -63,7 +210,6 @@ export class MemoryAdapter { return true; }); - // Sort ascending by event timestamp results.sort((a, b) => new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime()); if (typeof limit === "number" && limit > 0) { @@ -73,6 +219,71 @@ export class MemoryAdapter { return results; } + /** + * Query 1-minute downsampled data for a room. + * @private + */ + _queryRoomDownsample(roomId, from, to, limit, _tier) { + let rows = this._downsample1m.filter((r) => { + if (r.roomId !== roomId) return false; + if (from instanceof Date && r.bucketEnd < from.getTime()) return false; + if (to instanceof Date && r.bucketStart > to.getTime()) return false; + return true; + }); + + rows.sort((a, b) => a.bucketStart - b.bucketStart); + + if (typeof limit === "number" && limit > 0) { + rows = rows.slice(0, limit); + } + + return rows.map((r) => ({ + clientId: "downsample", + roomId: r.roomId, + latitude: r.latitude, + longitude: r.longitude, + altitude: r.altitude, + accuracy: null, + speed: r.maxSpeed, + timestamp: new Date(r.bucketStart).toISOString(), + id: uuid(), + createdAt: new Date().toISOString(), + })); + } + + /** + * Query hourly or daily aggregate data for a room. + * @private + */ + _queryRoomAggregate(roomId, from, to, limit, tier) { + const source = tier === "1h" ? this._aggregate1h : this._aggregate1d; + let rows = source.filter((r) => { + if (r.roomId !== roomId) return false; + if (from instanceof Date && r.bucketEnd < from.getTime()) return false; + if (to instanceof Date && r.bucketStart > to.getTime()) return false; + return true; + }); + + rows.sort((a, b) => a.bucketStart - b.bucketStart); + + if (typeof limit === "number" && limit > 0) { + rows = rows.slice(0, limit); + } + + return rows.map((r) => ({ + clientId: `aggregate-${tier}`, + roomId: r.roomId, + latitude: r.latitude, + longitude: r.longitude, + altitude: r.altitude, + accuracy: null, + speed: r.avgSpeed, + timestamp: new Date(r.bucketStart).toISOString(), + id: uuid(), + createdAt: new Date().toISOString(), + })); + } + /** * Returns events whose coordinates fall within the bounding box. * O(n) linear scan — acceptable for testing; not for production scale. @@ -119,6 +330,270 @@ export class MemoryAdapter { ); } + /** + * Run in-memory compaction: delete expired raw events, compute + * 1-minute downsamples, 1-hour aggregates, and 1-day rollups. + * + * @param {object} [overrides={}] + * @param {number} [overrides.rawRetentionDays] + * @param {number} [overrides.downsample1mRetentionDays] + * @param {number} [overrides.downsample1hRetentionDays] + * @param {number} [overrides.aggregate1dRetentionDays] + * @returns {Promise} + */ + async compact(overrides = {}) { + if (this._closed) throw new Error("MemoryAdapter is closed"); + + const start = Date.now(); + const retention = { ...this._retention, ...overrides }; + let rawDeleted = 0; + + // Phase 1: Delete expired raw events + const rawCutoff = Date.now() - retention.rawRetentionDays * 86400000; + const beforeCount = this._events.length; + this._events = this._events.filter((e) => new Date(e.timestamp).getTime() >= rawCutoff); + rawDeleted = beforeCount - this._events.length; + + // Phase 2: Compute 1-minute downsamples + const downsample1m = this._computeDownsample1m(); + this._downsample1m = downsample1m; + + // Phase 3: Compute 1-hour aggregates from 1m data + const aggregate1h = this._computeAggregate1h(); + this._aggregate1h = aggregate1h; + + // Phase 4: Compute 1-day aggregates from 1h data + const aggregate1d = this._computeAggregate1d(); + this._aggregate1d = aggregate1d; + + // Phase 5: Trim downsample/aggregate tiers to retention + const m1Cutoff = Date.now() - retention.downsample1mRetentionDays * 86400000; + const h1Cutoff = Date.now() - retention.downsample1hRetentionDays * 86400000; + const d1Cutoff = Date.now() - retention.aggregate1dRetentionDays * 86400000; + this._downsample1m = this._downsample1m.filter((r) => r.bucketStart >= m1Cutoff); + this._aggregate1h = this._aggregate1h.filter((r) => r.bucketStart >= h1Cutoff); + this._aggregate1d = this._aggregate1d.filter((r) => r.bucketStart >= d1Cutoff); + + this._lastCompactionRun = new Date().toISOString(); + + return { + rawDeleted, + downsample1m: this._downsample1m.length, + aggregate1h: this._aggregate1h.length, + aggregate1d: this._aggregate1d.length, + durationMs: Date.now() - start, + }; + } + + /** + * Compute 1-minute downsample buckets from raw events. + * @private + * @returns {object[]} + */ + _computeDownsample1m() { + const buckets = new Map(); + for (const e of this._events) { + const key = `${e.roomId}|${truncateToMinute(e.timestamp)}`; + if (!buckets.has(key)) { + buckets.set(key, { + roomId: e.roomId, + bucketStart: truncateToMinute(e.timestamp), + lats: [], + lons: [], + alts: [], + speeds: [], + }); + } + const b = buckets.get(key); + b.lats.push(e.latitude); + b.lons.push(e.longitude); + if (e.altitude != null) b.alts.push(e.altitude); + if (e.speed != null) b.speeds.push(e.speed); + } + + return Array.from(buckets.values()).map((b) => ({ + roomId: b.roomId, + bucketStart: b.bucketStart, + bucketEnd: b.bucketStart + 59999, + latitude: median(b.lats), + longitude: median(b.lons), + altitude: b.alts.length > 0 ? median(b.alts) : null, + maxSpeed: b.speeds.length > 0 ? Math.max(...b.speeds) : null, + pointCount: b.lats.length, + })); + } + + /** + * Compute 1-hour aggregates from 1-minute downsample data. + * @private + * @returns {object[]} + */ + _computeAggregate1h() { + if (this._downsample1m.length === 0) return []; + + const buckets = new Map(); + for (const r of this._downsample1m) { + const key = `${r.roomId}|${truncateToHour(new Date(r.bucketStart))}`; + if (!buckets.has(key)) { + buckets.set(key, { + roomId: r.roomId, + bucketStart: truncateToHour(new Date(r.bucketStart)), + points: [], + totalPointCount: 0, + }); + } + const b = buckets.get(key); + b.points.push({ lat: r.latitude, lon: r.longitude }); + b.totalPointCount += r.pointCount; + } + + return Array.from(buckets.values()).map((b) => { + let totalDistance = 0; + for (let i = 1; i < b.points.length; i++) { + totalDistance += haversine( + b.points[i - 1].lat, b.points[i - 1].lon, + b.points[i].lat, b.points[i].lon + ); + } + const avgLat = b.points.reduce((s, p) => s + p.lat, 0) / b.points.length; + const avgLon = b.points.reduce((s, p) => s + p.lon, 0) / b.points.length; + + return { + roomId: b.roomId, + bucketStart: b.bucketStart, + bucketEnd: b.bucketStart + 3599999, + latitude: avgLat, + longitude: avgLon, + altitude: null, + avgSpeed: null, + maxSpeed: null, + minSpeed: null, + totalDistance, + pointCount: b.totalPointCount, + }; + }); + } + + /** + * Compute 1-day aggregates from 1-hour aggregate data. + * @private + * @returns {object[]} + */ + _computeAggregate1d() { + if (this._aggregate1h.length === 0) return []; + + const buckets = new Map(); + for (const r of this._aggregate1h) { + const key = `${r.roomId}|${truncateToDay(new Date(r.bucketStart))}`; + if (!buckets.has(key)) { + buckets.set(key, { + roomId: r.roomId, + bucketStart: truncateToDay(new Date(r.bucketStart)), + points: [], + totalPointCount: 0, + totalDistance: 0, + }); + } + const b = buckets.get(key); + b.points.push({ lat: r.latitude, lon: r.longitude }); + b.totalPointCount += r.pointCount; + b.totalDistance += r.totalDistance; + } + + return Array.from(buckets.values()).map((b) => { + const avgLat = b.points.reduce((s, p) => s + p.lat, 0) / b.points.length; + const avgLon = b.points.reduce((s, p) => s + p.lon, 0) / b.points.length; + + return { + roomId: b.roomId, + bucketStart: b.bucketStart, + bucketEnd: b.bucketStart + 86399999, + latitude: avgLat, + longitude: avgLon, + altitude: null, + avgSpeed: null, + maxSpeed: null, + minSpeed: null, + totalDistance: b.totalDistance, + pointCount: b.totalPointCount, + }; + }); + } + + /** + * Return compaction status with per-tier statistics. + * + * @returns {Promise} + */ + async getCompactionStatus() { + if (this._closed) throw new Error("MemoryAdapter is closed"); + + const rawOldest = this._events.length > 0 + ? new Date(Math.min(...this._events.map((e) => new Date(e.timestamp).getTime()))).toISOString() + : null; + const rawNewest = this._events.length > 0 + ? new Date(Math.max(...this._events.map((e) => new Date(e.timestamp).getTime()))).toISOString() + : null; + + const m1Oldest = this._downsample1m.length > 0 + ? new Date(Math.min(...this._downsample1m.map((r) => r.bucketStart))).toISOString() + : null; + const m1Newest = this._downsample1m.length > 0 + ? new Date(Math.max(...this._downsample1m.map((r) => r.bucketStart))).toISOString() + : null; + + const h1Oldest = this._aggregate1h.length > 0 + ? new Date(Math.min(...this._aggregate1h.map((r) => r.bucketStart))).toISOString() + : null; + const h1Newest = this._aggregate1h.length > 0 + ? new Date(Math.max(...this._aggregate1h.map((r) => r.bucketStart))).toISOString() + : null; + + const d1Oldest = this._aggregate1d.length > 0 + ? new Date(Math.min(...this._aggregate1d.map((r) => r.bucketStart))).toISOString() + : null; + const d1Newest = this._aggregate1d.length > 0 + ? new Date(Math.max(...this._aggregate1d.map((r) => r.bucketStart))).toISOString() + : null; + + return { + lastRun: this._lastCompactionRun, + nextRun: this._nextCompactionRun + ? new Date(this._nextCompactionRun).toISOString() + : null, + tiers: [ + { + name: "raw", + rows: this._events.length, + sizeBytes: JSON.stringify(this._events).length, + oldest: rawOldest, + newest: rawNewest, + }, + { + name: "1m", + rows: this._downsample1m.length, + sizeBytes: JSON.stringify(this._downsample1m).length, + oldest: m1Oldest, + newest: m1Newest, + }, + { + name: "1h", + rows: this._aggregate1h.length, + sizeBytes: JSON.stringify(this._aggregate1h).length, + oldest: h1Oldest, + newest: h1Newest, + }, + { + name: "1d", + rows: this._aggregate1d.length, + sizeBytes: JSON.stringify(this._aggregate1d).length, + oldest: d1Oldest, + newest: d1Newest, + }, + ], + }; + } + /** * Clears internal state. Idempotent. * @@ -127,6 +602,9 @@ export class MemoryAdapter { async close() { this._closed = true; this._events = []; + this._downsample1m = []; + this._aggregate1h = []; + this._aggregate1d = []; } /** @@ -137,5 +615,8 @@ export class MemoryAdapter { */ clear() { this._events = []; + this._downsample1m = []; + this._aggregate1h = []; + this._aggregate1d = []; } } diff --git a/src/storage/migration-compaction.sql b/src/storage/migration-compaction.sql new file mode 100644 index 0000000..d7d537a --- /dev/null +++ b/src/storage/migration-compaction.sql @@ -0,0 +1,279 @@ +-- Migration: Time-Series Compaction for location_events +-- Issue #257: Implement time-series data compaction with automated downsampling, +-- retention policies, and continuous aggregates. +-- +-- This migration: +-- 1. Migrates location_events to a partitioned table (by day on timestamp) +-- 2. Creates 1-minute downsample, 1-hour aggregate, and 1-day aggregate tables +-- 3. Creates helper functions for compaction and querying +-- +-- Prerequisites: +-- - PostgreSQL 16+ (for declarative partitioning) +-- - Optional: TimescaleDB extension for continuous aggregates +-- +-- Usage: +-- psql -d spatial_tracking -f migration-compaction.sql + +-- ============================================================ +-- 0. Optionally enable TimescaleDB (skip if not installed) +-- ============================================================ +-- CREATE EXTENSION IF NOT EXISTS timescaledb; + +-- ============================================================ +-- 1. Migrate location_events to partitioned table +-- ============================================================ + +-- Create new partitioned table +CREATE TABLE IF NOT EXISTS location_events_new ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + client_id TEXT NOT NULL, + room_id TEXT NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + accuracy DOUBLE PRECISION, + speed DOUBLE PRECISION, + timestamp TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +) PARTITION BY RANGE (timestamp); + +-- Migrate data if old table exists and is not yet partitioned +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_name = 'location_events' + AND table_type = 'BASE TABLE' + ) AND NOT EXISTS ( + SELECT 1 FROM pg_class + WHERE relname = 'location_events' + AND relkind = 'p' + ) THEN + -- Copy data to new partitioned table + INSERT INTO location_events_new + SELECT id, client_id, room_id, latitude, longitude, altitude, accuracy, speed, timestamp, created_at + FROM location_events; + + -- Drop old table and rename new + DROP TABLE location_events; + ALTER TABLE location_events_new RENAME TO location_events; + ELSIF NOT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_name = 'location_events' + ) THEN + -- No existing table, just rename the new one + ALTER TABLE location_events_new RENAME TO location_events; + ELSE + -- Already partitioned, drop the unused new table + DROP TABLE IF EXISTS location_events_new; + END IF; +END $$; + +-- Create indexes on the partitioned table +CREATE INDEX IF NOT EXISTS location_events_room_id_timestamp_idx + ON location_events (room_id, timestamp DESC); + +CREATE INDEX IF NOT EXISTS location_events_lat_lon_idx + ON location_events (latitude, longitude); + +-- ============================================================ +-- 2. Create partition creation helper +-- ============================================================ + +CREATE OR REPLACE FUNCTION create_daily_partition(partition_date DATE) +RETURNS VOID AS $$ +DECLARE + partition_name TEXT; + start_date DATE; + end_date DATE; +BEGIN + partition_name := 'location_events_' || to_char(partition_date, 'YYYY_MM_DD'); + start_date := partition_date; + end_date := partition_date + INTERVAL '1 day'; + + -- Only create if it does not exist + IF NOT EXISTS ( + SELECT 1 FROM pg_class WHERE relname = partition_name + ) THEN + EXECUTE format( + 'CREATE TABLE %I PARTITION OF location_events FOR VALUES FROM (%L) TO (%L)', + partition_name, start_date, end_date + ); + END IF; +END; +$$ LANGUAGE plpgsql; + +-- ============================================================ +-- 3. Auto-create partitions for the next 30 days +-- ============================================================ + +DO $$ +DECLARE + i INT; + d DATE; +BEGIN + FOR i IN 0..30 LOOP + d := CURRENT_DATE + (i || ' days')::INTERVAL; + PERFORM create_daily_partition(d); + END LOOP; +END $$; + +-- ============================================================ +-- 4. Create 1-minute downsample table +-- ============================================================ + +CREATE TABLE IF NOT EXISTS location_events_1m ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + room_id TEXT NOT NULL, + bucket_start TIMESTAMPTZ NOT NULL, + bucket_end TIMESTAMPTZ NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + max_speed DOUBLE PRECISION, + point_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (room_id, bucket_start) +); + +CREATE INDEX IF NOT EXISTS idx_1m_room_bucket + ON location_events_1m (room_id, bucket_start DESC); + +-- ============================================================ +-- 5. Create 1-hour aggregate table +-- ============================================================ + +CREATE TABLE IF NOT EXISTS location_events_1h ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + room_id TEXT NOT NULL, + bucket_start TIMESTAMPTZ NOT NULL, + bucket_end TIMESTAMPTZ NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + avg_speed DOUBLE PRECISION, + max_speed DOUBLE PRECISION, + min_speed DOUBLE PRECISION, + total_distance DOUBLE PRECISION DEFAULT 0, + point_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (room_id, bucket_start) +); + +CREATE INDEX IF NOT EXISTS idx_1h_room_bucket + ON location_events_1h (room_id, bucket_start DESC); + +-- ============================================================ +-- 6. Create 1-day aggregate table +-- ============================================================ + +CREATE TABLE IF NOT EXISTS location_events_1d ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + room_id TEXT NOT NULL, + bucket_start TIMESTAMPTZ NOT NULL, + bucket_end TIMESTAMPTZ NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + avg_speed DOUBLE PRECISION, + max_speed DOUBLE PRECISION, + min_speed DOUBLE PRECISION, + total_distance DOUBLE PRECISION DEFAULT 0, + point_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (room_id, bucket_start) +); + +CREATE INDEX IF NOT EXISTS idx_1d_room_bucket + ON location_events_1d (room_id, bucket_start DESC); + +-- ============================================================ +-- 7. Create partition pruning helper for old raw data +-- ============================================================ + +CREATE OR REPLACE FUNCTION drop_old_raw_partitions(retention_days INT) +RETURNS TABLE(dropped_partition TEXT, dropped_rows BIGINT) AS $$ +DECLARE + r RECORD; + cutoff_date DATE; + part_date DATE; + row_count BIGINT; +BEGIN + cutoff_date := CURRENT_DATE - (retention_days || ' days')::INTERVAL; + + FOR r IN + SELECT inhrelid::regclass::text AS partition_name + FROM pg_inherits + JOIN pg_class parent ON pg_inherits.inhparent = parent.oid + JOIN pg_class child ON pg_inherits.inhrelid = child.oid + WHERE parent.relname = 'location_events' + AND child.relname LIKE 'location_events_%\_%\_%\_%\_%\_%' + LOOP + -- Extract date from partition name + BEGIN + part_date := to_date( + replace(r.partition_name, 'location_events_', ''), + 'YYYY_MM_DD' + ); + + IF part_date < cutoff_date THEN + -- Count rows before dropping + EXECUTE format('SELECT count(*) FROM %I', r.partition_name) INTO row_count; + + -- Detach and drop the partition + EXECUTE format( + 'ALTER TABLE location_events DETACH PARTITION %I', + r.partition_name + ); + EXECUTE format('DROP TABLE %I', r.partition_name); + + dropped_partition := r.partition_name; + dropped_rows := row_count; + RETURN NEXT; + END IF; + EXCEPTION WHEN OTHERS THEN + -- Skip partitions with non-standard naming + CONTINUE; + END; + END LOOP; +END; +$$ LANGUAGE plpgsql; + +-- ============================================================ +-- 8. Create partition auto-creation trigger +-- ============================================================ + +CREATE OR REPLACE FUNCTION auto_create_partition_trigger() +RETURNS TRIGGER AS $$ +DECLARE + partition_date DATE; +BEGIN + partition_date := date_trunc('day', NEW.timestamp)::DATE; + PERFORM create_daily_partition(partition_date); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Drop existing trigger if it exists +DROP TRIGGER IF EXISTS auto_partition_trigger ON location_events; + +CREATE TRIGGER auto_partition_trigger + BEFORE INSERT ON location_events + FOR EACH ROW + EXECUTE FUNCTION auto_create_partition_trigger(); + +-- ============================================================ +-- 9. Create compaction status tracking table +-- ============================================================ + +CREATE TABLE IF NOT EXISTS compaction_status ( + id INTEGER PRIMARY KEY DEFAULT 1, + last_run TIMESTAMPTZ, + next_run TIMESTAMPTZ, + last_result JSONB, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +INSERT INTO compaction_status (id, last_run, next_run, updated_at) +VALUES (1, NULL, NULL, NOW()) +ON CONFLICT (id) DO NOTHING; diff --git a/src/storage/postgres.js b/src/storage/postgres.js index 6c50925..b9a0738 100644 --- a/src/storage/postgres.js +++ b/src/storage/postgres.js @@ -3,7 +3,7 @@ * * Features: * - Connection pooling via `pg` Pool (dynamic import so `pg` is optional). - * - Auto-creates the `location_events` table and indexes on first use. + * - Auto-creates partitioned `location_events` table and indexes on first use. * - Write batching: events accumulate in a bounded in-memory buffer and are * flushed either when `batchSize` is reached or every `flushIntervalMs`. * - Bulk insert via `INSERT … SELECT * FROM unnest(…)` — one round-trip per batch. @@ -11,18 +11,25 @@ * are dropped and `storage_dropped_events_total` is incremented. * - Spatial index: composite (latitude, longitude) GiST-capable index for * bounding-box queries without requiring PostGIS. + * - Time-series compaction: automated downsampling, retention policies, and + * continuous aggregates for cost-effective long-term historical storage. + * - Declarative partitioning by day for efficient data lifecycle management. + * - Transparent query routing: queryRoom supports resolution parameter + * ("raw" | "1m" | "1h" | "1d" | "auto") for tiered data access. */ import { logger } from "../logger.js"; +// ─────────────────────────── Schema DDL ──────────────────────────────────────── + /** - * DDL executed once on startup to ensure the schema exists. - * Uses DOUBLE PRECISION for coordinates and TIMESTAMPTZ for temporal columns - * as required by the issue specification. + * DDL for the partitioned location_events table. + * Uses DOUBLE PRECISION for coordinates and TIMESTAMPTZ for temporal columns. + * Partitioned by day on the timestamp column for efficient lifecycle management. */ const CREATE_TABLE_SQL = ` CREATE TABLE IF NOT EXISTS location_events ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + id UUID NOT NULL DEFAULT gen_random_uuid(), client_id TEXT NOT NULL, room_id TEXT NOT NULL, latitude DOUBLE PRECISION NOT NULL, @@ -32,7 +39,7 @@ CREATE TABLE IF NOT EXISTS location_events ( speed DOUBLE PRECISION, timestamp TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); +) PARTITION BY RANGE (timestamp); `; const CREATE_INDEXES_SQL = ` @@ -45,10 +52,7 @@ CREATE INDEX IF NOT EXISTS location_events_lat_lon_idx /** * Bulk insert using unnest — single round-trip for an arbitrarily large batch. - * - * Parameter arrays are positionally matched: - * $1 = client_id[], $2 = room_id[], $3 = latitude[], $4 = longitude[], - * $5 = altitude[], $6 = accuracy[], $7 = speed[], $8 = timestamp[] + * Parameter arrays are positionally matched. */ const BULK_INSERT_SQL = ` INSERT INTO location_events @@ -65,17 +69,358 @@ SELECT * FROM unnest( ) `; +// ─────────────────────────── Downsample DDL ──────────────────────────────────── + +const CREATE_DOWNSAMPLE_1M_SQL = ` +CREATE TABLE IF NOT EXISTS location_events_1m ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + room_id TEXT NOT NULL, + bucket_start TIMESTAMPTZ NOT NULL, + bucket_end TIMESTAMPTZ NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + max_speed DOUBLE PRECISION, + point_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (room_id, bucket_start) +)`; + +const CREATE_DOWNSAMPLE_1M_INDEX_SQL = ` +CREATE INDEX IF NOT EXISTS idx_1m_room_bucket + ON location_events_1m (room_id, bucket_start DESC) +`; + +const CREATE_AGGREGATE_1H_SQL = ` +CREATE TABLE IF NOT EXISTS location_events_1h ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + room_id TEXT NOT NULL, + bucket_start TIMESTAMPTZ NOT NULL, + bucket_end TIMESTAMPTZ NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + avg_speed DOUBLE PRECISION, + max_speed DOUBLE PRECISION, + min_speed DOUBLE PRECISION, + total_distance DOUBLE PRECISION DEFAULT 0, + point_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (room_id, bucket_start) +)`; + +const CREATE_AGGREGATE_1H_INDEX_SQL = ` +CREATE INDEX IF NOT EXISTS idx_1h_room_bucket + ON location_events_1h (room_id, bucket_start DESC) +`; + +const CREATE_AGGREGATE_1D_SQL = ` +CREATE TABLE IF NOT EXISTS location_events_1d ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + room_id TEXT NOT NULL, + bucket_start TIMESTAMPTZ NOT NULL, + bucket_end TIMESTAMPTZ NOT NULL, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + altitude DOUBLE PRECISION, + avg_speed DOUBLE PRECISION, + max_speed DOUBLE PRECISION, + min_speed DOUBLE PRECISION, + total_distance DOUBLE PRECISION DEFAULT 0, + point_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (room_id, bucket_start) +)`; + +const CREATE_AGGREGATE_1D_INDEX_SQL = ` +CREATE INDEX IF NOT EXISTS idx_1d_room_bucket + ON location_events_1d (room_id, bucket_start DESC) +`; + +// ─────────────────────────── Partition Helpers ────────────────────────────────── + +const CREATE_PARTITION_FUNCTION_SQL = ` +CREATE OR REPLACE FUNCTION create_daily_partition(partition_date DATE) +RETURNS VOID AS $func$ +DECLARE + partition_name TEXT; + start_date DATE; + end_date DATE; +BEGIN + partition_name := 'location_events_' || to_char(partition_date, 'YYYY_MM_DD'); + start_date := partition_date; + end_date := partition_date + INTERVAL '1 day'; + + IF NOT EXISTS ( + SELECT 1 FROM pg_class WHERE relname = partition_name + ) THEN + EXECUTE format( + 'CREATE TABLE %I PARTITION OF location_events FOR VALUES FROM (%L) TO (%L)', + partition_name, start_date, end_date + ); + END IF; +END; +$func$ LANGUAGE plpgsql; +`; + +const CREATE_PARTITION_TRIGGER_SQL = ` +CREATE OR REPLACE FUNCTION auto_create_partition_trigger() +RETURNS TRIGGER AS $func$ +DECLARE + partition_date DATE; +BEGIN + partition_date := date_trunc('day', NEW.timestamp)::DATE; + PERFORM create_daily_partition(partition_date); + RETURN NEW; +END; +$func$ LANGUAGE plpgsql; +`; + +const DROP_EXISTING_TRIGGER_SQL = `DROP TRIGGER IF EXISTS auto_partition_trigger ON location_events;`; + +const CREATE_PARTITION_TRIGGER_BIND_SQL = ` +CREATE TRIGGER auto_partition_trigger + BEFORE INSERT ON location_events + FOR EACH ROW + EXECUTE FUNCTION auto_create_partition_trigger(); +`; + +const CREATE_COMPACT_STATUS_SQL = ` +CREATE TABLE IF NOT EXISTS compaction_status ( + id INTEGER PRIMARY KEY DEFAULT 1, + last_run TIMESTAMPTZ, + next_run TIMESTAMPTZ, + last_result JSONB, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +`; + +// ─────────────────────────── Compaction Queries ──────────────────────────────── + +/** + * Compute 1-minute downsample rows from raw data for a given time range. + * Uses median for lat/lon, max for speed. + */ +const DOWNSAMPLE_1M_SQL = ` +INSERT INTO location_events_1m + (room_id, bucket_start, bucket_end, latitude, longitude, altitude, max_speed, point_count) +SELECT + room_id, + date_trunc('minute', timestamp) AS bucket_start, + date_trunc('minute', timestamp) + INTERVAL '59 seconds 999 milliseconds' AS bucket_end, + percentile_cont(0.5) WITHIN GROUP (ORDER BY latitude) AS latitude, + percentile_cont(0.5) WITHIN GROUP (ORDER BY longitude) AS longitude, + AVG(altitude) AS altitude, + MAX(speed) AS max_speed, + COUNT(*) AS point_count +FROM location_events +WHERE timestamp >= date_trunc('minute', $1::timestamptz) + AND timestamp < date_trunc('minute', $2::timestamptz) + INTERVAL '1 minute' +GROUP BY room_id, date_trunc('minute', timestamp) +ON CONFLICT (room_id, bucket_start) DO UPDATE SET + latitude = EXCLUDED.latitude, + longitude = EXCLUDED.longitude, + altitude = EXCLUDED.altitude, + max_speed = EXCLUDED.max_speed, + point_count = EXCLUDED.point_count, + bucket_end = EXCLUDED.bucket_end; +`; + +/** + * Compute 1-hour aggregates from 1-minute downsample data. + * Includes haversine distance between consecutive median points. + */ +const AGGREGATE_1H_SQL = ` +WITH hourly AS ( + SELECT + room_id, + date_trunc('hour', bucket_start) AS bucket_start, + AVG(latitude) AS latitude, + AVG(longitude) AS longitude, + AVG(altitude) AS altitude, + AVG(max_speed) AS avg_speed, + MAX(max_speed) AS max_speed, + MIN(max_speed) AS min_speed, + SUM(point_count) AS point_count + FROM location_events_1m + WHERE bucket_start >= date_trunc('hour', $1::timestamptz) + AND bucket_start < date_trunc('hour', $2::timestamptz) + INTERVAL '1 hour' + GROUP BY room_id, date_trunc('hour', bucket_start) +), +ordered AS ( + SELECT + h.*, + LAG(latitude) OVER (PARTITION BY room_id ORDER BY bucket_start) AS prev_lat, + LAG(longitude) OVER (PARTITION BY room_id ORDER BY bucket_start) AS prev_lon + FROM hourly h +) +INSERT INTO location_events_1h + (room_id, bucket_start, bucket_end, latitude, longitude, altitude, + avg_speed, max_speed, min_speed, total_distance, point_count) +SELECT + room_id, + bucket_start, + bucket_start + INTERVAL '59 minutes 59 seconds 999 milliseconds' AS bucket_end, + latitude, + longitude, + altitude, + avg_speed, + max_speed, + min_speed, + COALESCE( + (SELECT SUM( + 6371000.0 * 2 * asin(sqrt( + power(sin((o.latitude - o.prev_lat) / 2.0), 2) + + cos(radians(o.prev_lat)) * cos(radians(o.latitude)) * + power(sin((o.longitude - o.prev_lon) / 2.0), 2) + )) + ) FROM ordered o + WHERE o.room_id = ordered.room_id + AND o.bucket_start <= ordered.bucket_start + AND o.prev_lat IS NOT NULL + ), 0 + ) AS total_distance, + point_count +FROM ordered +ON CONFLICT (room_id, bucket_start) DO UPDATE SET + latitude = EXCLUDED.latitude, + longitude = EXCLUDED.longitude, + altitude = EXCLUDED.altitude, + avg_speed = EXCLUDED.avg_speed, + max_speed = EXCLUDED.max_speed, + min_speed = EXCLUDED.min_speed, + total_distance = EXCLUDED.total_distance, + point_count = EXCLUDED.point_count, + bucket_end = EXCLUDED.bucket_end; +`; + +/** + * Compute 1-day aggregates from 1-hour aggregate data. + */ +const AGGREGATE_1D_SQL = ` +INSERT INTO location_events_1d + (room_id, bucket_start, bucket_end, latitude, longitude, altitude, + avg_speed, max_speed, min_speed, total_distance, point_count) +SELECT + room_id, + date_trunc('day', bucket_start) AS bucket_start, + date_trunc('day', bucket_start) + INTERVAL '23 hours 59 minutes 59 seconds 999 milliseconds' AS bucket_end, + AVG(latitude) AS latitude, + AVG(longitude) AS longitude, + AVG(altitude) AS altitude, + AVG(avg_speed) AS avg_speed, + MAX(max_speed) AS max_speed, + MIN(min_speed) AS min_speed, + SUM(total_distance) AS total_distance, + SUM(point_count) AS point_count +FROM location_events_1h +WHERE bucket_start >= date_trunc('day', $1::timestamptz) + AND bucket_start < date_trunc('day', $2::timestamptz) + INTERVAL '1 day' +GROUP BY room_id, date_trunc('day', bucket_start) +ON CONFLICT (room_id, bucket_start) DO UPDATE SET + latitude = EXCLUDED.latitude, + longitude = EXCLUDED.longitude, + altitude = EXCLUDED.altitude, + avg_speed = EXCLUDED.avg_speed, + max_speed = EXCLUDED.max_speed, + min_speed = EXCLUDED.min_speed, + total_distance = EXCLUDED.total_distance, + point_count = EXCLUDED.point_count, + bucket_end = EXCLUDED.bucket_end; +`; + +const CREATE_DROP_PARTITION_FUNCTION_SQL = ` +CREATE OR REPLACE FUNCTION drop_old_raw_partitions(retention_days INT) +RETURNS TABLE(dropped_partition TEXT, dropped_rows BIGINT) AS $func$ +DECLARE + r RECORD; + cutoff_date DATE; + part_date DATE; + row_count BIGINT; +BEGIN + cutoff_date := CURRENT_DATE - (retention_days || ' days')::INTERVAL; + + FOR r IN + SELECT child.relname AS partition_name + FROM pg_inherits + JOIN pg_class parent ON pg_inherits.inhparent = parent.oid + JOIN pg_class child ON pg_inherits.inhrelid = child.oid + WHERE parent.relname = 'location_events' + AND child.relkind = 'r' + LOOP + BEGIN + part_date := to_date( + replace(r.partition_name, 'location_events_', ''), + 'YYYY_MM_DD' + ); + + IF part_date < cutoff_date THEN + EXECUTE format('SELECT count(*) FROM %I', r.partition_name) INTO row_count; + EXECUTE format('ALTER TABLE location_events DETACH PARTITION %I', r.partition_name); + EXECUTE format('DROP TABLE %I', r.partition_name); + dropped_partition := r.partition_name; + dropped_rows := row_count; + RETURN NEXT; + END IF; + EXCEPTION WHEN OTHERS THEN + CONTINUE; + END; + END LOOP; +END; +$func$ LANGUAGE plpgsql; +`; + +// ─────────────────────────── Tier trim queries ───────────────────────────────── + +const TRIM_TIER_1M_SQL = `DELETE FROM location_events_1m WHERE bucket_start < $1::timestamptz;`; +const TRIM_TIER_1H_SQL = `DELETE FROM location_events_1h WHERE bucket_start < $1::timestamptz;`; +const TRIM_TIER_1D_SQL = `DELETE FROM location_events_1d WHERE bucket_start < $1::timestamptz;`; + +// ─────────────────────────── Status queries ──────────────────────────────────── + +const GET_COMPACT_STATUS_SQL = `SELECT last_run, next_run, last_result FROM compaction_status WHERE id = 1;`; +const UPDATE_COMPACT_STATUS_SQL = ` +UPDATE compaction_status SET last_run = $1, next_run = $2, last_result = $3::jsonb, updated_at = NOW() WHERE id = 1; +`; + +const RAW_TIER_STATS_SQL = ` +SELECT + 'raw' AS tier_name, + COALESCE( + (SELECT SUM(c.reltuples)::bigint FROM pg_inherits i + JOIN pg_class parent ON i.inhparent = parent.oid + JOIN pg_class c ON i.inhrelid = c.oid + WHERE parent.relname = 'location_events'), + 0 + ) AS row_count, + COALESCE( + (SELECT pg_total_relation_size(parent.oid) FROM pg_class parent WHERE parent.relname = 'location_events'), + 0 + ) AS size_bytes, + (SELECT MIN(timestamp) FROM location_events) AS oldest, + (SELECT MAX(timestamp) FROM location_events) AS newest; +`; + +// ─────────────────────────── Class ───────────────────────────────────────────── + /** * @implements {import("./adapter.js").StorageAdapter} */ export class PostgresAdapter { /** * @param {object} [config={}] - * @param {string} [config.connectionString] - Postgres connection URL. Falls back to DATABASE_URL. - * @param {number} [config.poolSize=10] - Maximum pool connections. - * @param {number} [config.batchSize=100] - Flush when buffer reaches this size. - * @param {number} [config.flushIntervalMs=1000]- Flush every N milliseconds regardless of size. - * @param {number} [config.maxBufferSize=10000] - Hard cap on in-flight buffer; oldest dropped beyond this. + * @param {string} [config.connectionString] - Postgres connection URL. Falls back to DATABASE_URL. + * @param {number} [config.poolSize=10] - Maximum pool connections. + * @param {number} [config.batchSize=100] - Flush when buffer reaches this size. + * @param {number} [config.flushIntervalMs=1000] - Flush every N milliseconds regardless of size. + * @param {number} [config.maxBufferSize=10000] - Hard cap on in-flight buffer; oldest dropped beyond this. + * @param {number} [config.rawRetentionDays=7] - Days to keep raw events. + * @param {number} [config.downsample1mRetentionDays=90] - Days to keep 1m downsample. + * @param {number} [config.downsample1hRetentionDays=365] - Days to keep 1h aggregates. + * @param {number} [config.aggregate1dRetentionDays=2555] - Days to keep 1d rollups. + * @param {number} [config.compactionIntervalMs=300000] - Compaction interval (default 5 min). + * @param {number} [config.compactionBatchSize=10000] - Batch size for deletion loops. + * @param {boolean} [config.timescaleDbEnabled=false] - Use TimescaleDB continuous aggregates. */ constructor(config = {}) { this._connectionString = @@ -85,6 +430,17 @@ export class PostgresAdapter { this._flushIntervalMs = config.flushIntervalMs ?? 1000; this._maxBufferSize = config.maxBufferSize ?? 10000; + // Retention configuration + this._rawRetentionDays = config.rawRetentionDays ?? 7; + this._downsample1mRetentionDays = config.downsample1mRetentionDays ?? 90; + this._downsample1hRetentionDays = config.downsample1hRetentionDays ?? 365; + this._aggregate1dRetentionDays = config.aggregate1dRetentionDays ?? 2555; + + // Compaction configuration + this._compactionIntervalMs = config.compactionIntervalMs ?? 300000; + this._compactionBatchSize = config.compactionBatchSize ?? 10000; + this._timescaleDbEnabled = config.timescaleDbEnabled ?? false; + /** @type {import("./adapter.js").LocationEvent[]} */ this._buffer = []; @@ -97,6 +453,9 @@ export class PostgresAdapter { /** @type {NodeJS.Timeout|null} */ this._flushTimer = null; + /** @type {NodeJS.Timeout|null} */ + this._compactionTimer = null; + /** Promise that resolves when the schema has been initialised. */ this._initPromise = null; @@ -105,6 +464,18 @@ export class PostgresAdapter { /** Whether a flush is currently in-flight (prevents double-flush). */ this._flushing = false; + + /** Whether compaction is currently running (prevents double-run). */ + this._compacting = false; + + /** Last compaction run result. */ + this._lastCompactionResult = null; + + /** Last compaction run timestamp. */ + this._lastCompactionRun = null; + + /** Next scheduled compaction run timestamp. */ + this._nextCompactionRun = null; } // ─────────────────────────── lifecycle ──────────────────────────────────── @@ -122,7 +493,6 @@ export class PostgresAdapter { } async _doInit() { - // Dynamic import keeps pg optional when the memory adapter is used. const { default: pg } = await import("pg"); const { Pool } = pg; @@ -131,27 +501,67 @@ export class PostgresAdapter { max: this._poolSize, }); - // Surface connection errors without crashing the process. this._pool.on("error", (err) => { logger.error("PostgresAdapter pool error", { error: err.message }); }); - // Run schema migration. + // Run schema migrations const client = await this._pool.connect(); try { + // 1. Partitioned location_events table await client.query(CREATE_TABLE_SQL); await client.query(CREATE_INDEXES_SQL); + + // 2. Partition helper functions + await client.query(CREATE_PARTITION_FUNCTION_SQL); + await client.query(CREATE_DROP_PARTITION_FUNCTION_SQL); + + // 3. Partition trigger + await client.query(CREATE_PARTITION_TRIGGER_SQL); + await client.query(DROP_EXISTING_TRIGGER_SQL); + await client.query(CREATE_PARTITION_TRIGGER_BIND_SQL); + + // 4. Create initial partitions (today + next 30 days) + await client.query(` + DO $$ + DECLARE + i INT; + d DATE; + BEGIN + FOR i IN 0..30 LOOP + d := CURRENT_DATE + (i || ' days')::INTERVAL; + PERFORM create_daily_partition(d); + END LOOP; + END $$; + `); + + // 5. Downsample and aggregate tables + await client.query(CREATE_DOWNSAMPLE_1M_SQL); + await client.query(CREATE_DOWNSAMPLE_1M_INDEX_SQL); + await client.query(CREATE_AGGREGATE_1H_SQL); + await client.query(CREATE_AGGREGATE_1H_INDEX_SQL); + await client.query(CREATE_AGGREGATE_1D_SQL); + await client.query(CREATE_AGGREGATE_1D_INDEX_SQL); + + // 6. Compaction status table + await client.query(CREATE_COMPACT_STATUS_SQL); + await client.query(` + INSERT INTO compaction_status (id, last_run, next_run, updated_at) + VALUES (1, NULL, NULL, NOW()) + ON CONFLICT (id) DO NOTHING; + `); } finally { client.release(); } - // Start the periodic flush timer. + // Start periodic flush timer this._flushTimer = setInterval(() => { this._scheduleFlush(); }, this._flushIntervalMs); - - // Node should not wait for the timer to exit. if (this._flushTimer.unref) this._flushTimer.unref(); + + // Start compaction timer + this._scheduleCompaction(); } // ─────────────────────────── write pipeline ─────────────────────────────── @@ -159,13 +569,6 @@ export class PostgresAdapter { /** * Accepts a batch of events into the write buffer. * - * This method is intentionally non-blocking: it enqueues events and returns - * immediately. The flush happens asynchronously so the WebSocket broadcast - * pipeline is never stalled by I/O. - * - * Backpressure enforcement: when the buffer exceeds `maxBufferSize`, the - * oldest events are evicted and counted in `storage_dropped_events_total`. - * * @param {import("./adapter.js").LocationEvent[]} events * @returns {Promise} */ @@ -173,15 +576,10 @@ export class PostgresAdapter { if (this._closed) throw new Error("PostgresAdapter is closed"); if (!Array.isArray(events) || events.length === 0) return; - // Ensure pool + schema are ready (no-op after first call). - // We do NOT await here so writeBatch returns immediately. - // The flush that actually writes to PG will await _init internally. this._ensureInit(); - // Append to buffer. this._buffer.push(...events); - // Enforce hard cap: drop oldest events if buffer is too large. if (this._buffer.length > this._maxBufferSize) { const excess = this._buffer.length - this._maxBufferSize; this._buffer.splice(0, excess); @@ -192,39 +590,24 @@ export class PostgresAdapter { }); } - // Flush immediately if we've hit the batch threshold. if (this._buffer.length >= this._batchSize) { this._scheduleFlush(); } } - /** - * Kicks off an async flush without blocking the caller. - * @private - */ _scheduleFlush() { - // Run flush async; do not propagate errors to caller. this._flush().catch((err) => { logger.error("PostgresAdapter flush error", { error: err.message }); }); } - /** - * Flushes the current buffer contents to Postgres. - * Concurrent flushes are serialised by the `_flushing` flag. - * - * @private - * @returns {Promise} - */ async _flush() { if (this._flushing || this._buffer.length === 0) return; this._flushing = true; try { - // Ensure the pool is ready before we try to write. await this._init(); - // Drain the buffer one batch at a time. while (this._buffer.length > 0) { const batch = this._buffer.splice(0, this._batchSize); await this._insertBatch(batch); @@ -234,40 +617,21 @@ export class PostgresAdapter { } } - /** - * Executes a single bulk INSERT for the given batch. - * - * @private - * @param {import("./adapter.js").LocationEvent[]} batch - * @returns {Promise} - */ async _insertBatch(batch) { const clientIds = batch.map((e) => e.clientId); - const roomIds = batch.map((e) => e.roomId); - const lats = batch.map((e) => e.latitude); - const lons = batch.map((e) => e.longitude); - const alts = batch.map((e) => e.altitude ?? null); - const accs = batch.map((e) => e.accuracy ?? null); - const speeds = batch.map((e) => e.speed ?? null); + const roomIds = batch.map((e) => e.roomId); + const lats = batch.map((e) => e.latitude); + const lons = batch.map((e) => e.longitude); + const alts = batch.map((e) => e.altitude ?? null); + const accs = batch.map((e) => e.accuracy ?? null); + const speeds = batch.map((e) => e.speed ?? null); const timestamps = batch.map((e) => e.timestamp); await this._pool.query(BULK_INSERT_SQL, [ - clientIds, - roomIds, - lats, - lons, - alts, - accs, - speeds, - timestamps, + clientIds, roomIds, lats, lons, alts, accs, speeds, timestamps, ]); } - /** - * Fires off the init promise without awaiting it, so the first writeBatch - * call does not block. - * @private - */ _ensureInit() { if (!this._initPromise) { this._init().catch((err) => { @@ -276,10 +640,296 @@ export class PostgresAdapter { } } + // ─────────────────────────── compaction pipeline ────────────────────────── + + /** + * Schedule periodic compaction runs. + * @private + */ + _scheduleCompaction() { + if (this._closed) return; + + this._nextCompactionRun = Date.now() + this._compactionIntervalMs; + + this._compactionTimer = setTimeout(async () => { + if (this._closed) return; + try { + await this.compact(); + } catch (err) { + logger.error("Compaction run failed", { error: err.message }); + } + this._scheduleCompaction(); + }, this._compactionIntervalMs); + + if (this._compactionTimer.unref) this._compactionTimer.unref(); + } + + /** + * Run the full compaction pipeline: + * 1. Drop expired raw partitions (instant DDL). + * 2. Compute 1-minute downsamples for the last 24 hours. + * 3. Compute 1-hour aggregates from new 1m data. + * 4. Compute 1-day aggregates from new 1h data. + * 5. Trim each tier to its retention window. + * + * Compaction is idempotent — re-running for the same time window + * performs upserts (ON CONFLICT DO UPDATE). + * + * @param {object} [overrides={}] + * @returns {Promise} + */ + async compact(overrides = {}) { + if (this._closed) throw new Error("PostgresAdapter is closed"); + if (this._compacting) { + logger.info("Compaction already in progress, skipping"); + return { + rawDeleted: 0, downsample1m: 0, aggregate1h: 0, + aggregate1d: 0, durationMs: 0, error: "already running", + }; + } + + this._compacting = true; + const start = Date.now(); + + try { + await this._init(); + + const rawRetention = overrides.rawRetentionDays ?? this._rawRetentionDays; + const m1Retention = overrides.downsample1mRetentionDays ?? this._downsample1mRetentionDays; + const h1Retention = overrides.downsample1hRetentionDays ?? this._downsample1hRetentionDays; + const d1Retention = overrides.aggregate1dRetentionDays ?? this._aggregate1dRetentionDays; + + let rawDeleted = 0; + + // Phase 1: Drop expired raw partitions (instant DDL) + try { + const dropResult = await this._pool.query( + `SELECT drop_old_raw_partitions($1)`, + [rawRetention] + ); + if (dropResult.rows.length > 0) { + rawDeleted = dropResult.rows.reduce((sum, r) => sum + Number(r.drop_old_raw_partitions || 0), 0); + } + logger.info("Raw partition drop completed", { rawDeleted }); + } catch (err) { + logger.warn("Raw partition drop failed", { error: err.message }); + } + + // Phase 2: Compute 1-minute downsamples (last 24h window, safe for incremental) + let downsample1m = 0; + try { + const m1Start = new Date(Date.now() - 24 * 3600000); + const m1End = new Date(); + const m1Result = await this._pool.query(DOWNSAMPLE_1M_SQL, [m1Start, m1End]); + downsample1m = m1Result.rowCount ?? 0; + logger.info("1m downsample completed", { rows: downsample1m }); + } catch (err) { + logger.warn("1m downsample failed", { error: err.message }); + } + + // Phase 3: Compute 1-hour aggregates (last 7 days window) + let aggregate1h = 0; + try { + const h1Start = new Date(Date.now() - 7 * 86400000); + const h1End = new Date(); + const h1Result = await this._pool.query(AGGREGATE_1H_SQL, [h1Start, h1End]); + aggregate1h = h1Result.rowCount ?? 0; + logger.info("1h aggregate completed", { rows: aggregate1h }); + } catch (err) { + logger.warn("1h aggregate failed", { error: err.message }); + } + + // Phase 4: Compute 1-day aggregates (last 90 days window) + let aggregate1d = 0; + try { + const d1Start = new Date(Date.now() - 90 * 86400000); + const d1End = new Date(); + const d1Result = await this._pool.query(AGGREGATE_1D_SQL, [d1Start, d1End]); + aggregate1d = d1Result.rowCount ?? 0; + logger.info("1d aggregate completed", { rows: aggregate1d }); + } catch (err) { + logger.warn("1d aggregate failed", { error: err.message }); + } + + // Phase 5: Trim each tier to retention window + try { + const m1Cutoff = new Date(Date.now() - m1Retention * 86400000); + const h1Cutoff = new Date(Date.now() - h1Retention * 86400000); + const d1Cutoff = new Date(Date.now() - d1Retention * 86400000); + await this._pool.query(TRIM_TIER_1M_SQL, [m1Cutoff]); + await this._pool.query(TRIM_TIER_1H_SQL, [h1Cutoff]); + await this._pool.query(TRIM_TIER_1D_SQL, [d1Cutoff]); + logger.info("Tier retention trimming completed"); + } catch (err) { + logger.warn("Tier retention trimming failed", { error: err.message }); + } + + const durationMs = Date.now() - start; + const result = { + rawDeleted, + downsample1m, + aggregate1h, + aggregate1d, + durationMs, + }; + + this._lastCompactionResult = result; + this._lastCompactionRun = new Date().toISOString(); + + // Update compaction status + try { + await this._pool.query(UPDATE_COMPACT_STATUS_SQL, [ + this._lastCompactionRun, + new Date(Date.now() + this._compactionIntervalMs).toISOString(), + JSON.stringify(result), + ]); + } catch (err) { + logger.warn("Failed to update compaction status", { error: err.message }); + } + + logger.info("Compaction completed", result); + return result; + } catch (err) { + const durationMs = Date.now() - start; + logger.error("Compaction failed", { error: err.message, durationMs }); + return { + rawDeleted: 0, downsample1m: 0, aggregate1h: 0, + aggregate1d: 0, durationMs, error: err.message, + }; + } finally { + this._compacting = false; + } + } + + /** + * Return current compaction status including last/next run times and + * per-tier statistics. + * + * @returns {Promise} + */ + async getCompactionStatus() { + if (this._closed) throw new Error("PostgresAdapter is closed"); + await this._init(); + + let lastRun = null; + let nextRun = null; + + try { + const statusResult = await this._pool.query(GET_COMPACT_STATUS_SQL); + if (statusResult.rows.length > 0) { + lastRun = statusResult.rows[0].last_run; + nextRun = statusResult.rows[0].next_run; + } + } catch (err) { + logger.warn("Failed to read compaction status", { error: err.message }); + } + + // Get tier stats + const tiers = []; + + // Raw tier + try { + const rawStats = await this._pool.query(RAW_TIER_STATS_SQL); + if (rawStats.rows.length > 0) { + const r = rawStats.rows[0]; + tiers.push({ + name: r.tier_name, + rows: Number(r.row_count), + sizeBytes: Number(r.size_bytes), + oldest: r.oldest ? new Date(r.oldest).toISOString() : null, + newest: r.newest ? new Date(r.newest).toISOString() : null, + }); + } + } catch { + tiers.push({ name: "raw", rows: 0, sizeBytes: 0, oldest: null, newest: null }); + } + + // 1m tier + try { + const m1Stats = await this._pool.query( + `SELECT '1m' AS tier_name, + COALESCE((SELECT count(*) FROM location_events_1m), 0) AS row_count, + COALESCE(pg_total_relation_size('location_events_1m'), 0) AS size_bytes, + (SELECT MIN(bucket_start) FROM location_events_1m) AS oldest, + (SELECT MAX(bucket_start) FROM location_events_1m) AS newest;` + ); + if (m1Stats.rows.length > 0) { + const r = m1Stats.rows[0]; + tiers.push({ + name: r.tier_name, + rows: Number(r.row_count), + sizeBytes: Number(r.size_bytes), + oldest: r.oldest ? new Date(r.oldest).toISOString() : null, + newest: r.newest ? new Date(r.newest).toISOString() : null, + }); + } + } catch { + tiers.push({ name: "1m", rows: 0, sizeBytes: 0, oldest: null, newest: null }); + } + + // 1h tier + try { + const h1Stats = await this._pool.query( + `SELECT '1h' AS tier_name, + COALESCE((SELECT count(*) FROM location_events_1h), 0) AS row_count, + COALESCE(pg_total_relation_size('location_events_1h'), 0) AS size_bytes, + (SELECT MIN(bucket_start) FROM location_events_1h) AS oldest, + (SELECT MAX(bucket_start) FROM location_events_1h) AS newest;` + ); + if (h1Stats.rows.length > 0) { + const r = h1Stats.rows[0]; + tiers.push({ + name: r.tier_name, + rows: Number(r.row_count), + sizeBytes: Number(r.size_bytes), + oldest: r.oldest ? new Date(r.oldest).toISOString() : null, + newest: r.newest ? new Date(r.newest).toISOString() : null, + }); + } + } catch { + tiers.push({ name: "1h", rows: 0, sizeBytes: 0, oldest: null, newest: null }); + } + + // 1d tier + try { + const d1Stats = await this._pool.query( + `SELECT '1d' AS tier_name, + COALESCE((SELECT count(*) FROM location_events_1d), 0) AS row_count, + COALESCE(pg_total_relation_size('location_events_1d'), 0) AS size_bytes, + (SELECT MIN(bucket_start) FROM location_events_1d) AS oldest, + (SELECT MAX(bucket_start) FROM location_events_1d) AS newest;` + ); + if (d1Stats.rows.length > 0) { + const r = d1Stats.rows[0]; + tiers.push({ + name: r.tier_name, + rows: Number(r.row_count), + sizeBytes: Number(r.size_bytes), + oldest: r.oldest ? new Date(r.oldest).toISOString() : null, + newest: r.newest ? new Date(r.newest).toISOString() : null, + }); + } + } catch { + tiers.push({ name: "1d", rows: 0, sizeBytes: 0, oldest: null, newest: null }); + } + + return { + lastRun: lastRun ? new Date(lastRun).toISOString() : null, + nextRun: nextRun ? new Date(nextRun).toISOString() : null, + tiers, + }; + } + // ─────────────────────────── read pipeline ──────────────────────────────── /** * Retrieves historical events for a room, ordered ascending by timestamp. + * Supports resolution parameter for tiered data access: + * "raw" - query raw location_events table + * "1m" - query 1-minute downsampled data + * "1h" - query 1-hour aggregate data + * "1d" - query 1-day aggregate data + * "auto" - select finest resolution fitting within limit (default) * * @param {string} roomId * @param {import("./adapter.js").QueryOptions} [options={}] @@ -289,7 +939,81 @@ export class PostgresAdapter { if (this._closed) throw new Error("PostgresAdapter is closed"); await this._init(); - const { from, to, limit } = options; + const { from, to, limit, resolution = "auto" } = options; + + let resolved = resolution; + if (resolution === "auto") { + resolved = await this._autoResolveResolution(roomId, from, to, limit); + } + + if (resolved === "1m") { + return this._queryRoom1m(roomId, from, to, limit); + } + if (resolved === "1h") { + return this._queryRoom1h(roomId, from, to, limit); + } + if (resolved === "1d") { + return this._queryRoom1d(roomId, from, to, limit); + } + + // Default: raw + return this._queryRoomRaw(roomId, from, to, limit); + } + + /** + * Auto-select the finest resolution that returns ≤ limit points. + * @private + */ + async _autoResolveResolution(roomId, from, to, limit) { + const maxPoints = limit || 1000; + const now = new Date(); + const fromMs = from ? from.getTime() : now.getTime() - 7 * 86400000; + const toMs = to ? to.getTime() : now.getTime(); + const rangeMs = toMs - fromMs; + + if (rangeMs <= 0) return "raw"; + + // Estimate raw points (1 per second) + const rawPointsEstimate = Math.floor(rangeMs / 1000); + if (rawPointsEstimate <= maxPoints) return "raw"; + + // Check 1m tier + const m1PointsEstimate = Math.floor(rangeMs / 60000); + if (m1PointsEstimate <= maxPoints) { + // Verify 1m data exists + try { + const result = await this._pool.query( + `SELECT count(*) FROM location_events_1m WHERE room_id = $1 AND bucket_start >= $2 AND bucket_start <= $3`, + [roomId, new Date(fromMs), new Date(toMs)] + ); + if (Number(result.rows[0].count) <= maxPoints) return "1m"; + } catch { + // 1m table may not have data, fall through + } + } + + // Check 1h tier + const h1PointsEstimate = Math.floor(rangeMs / 3600000); + if (h1PointsEstimate <= maxPoints) { + try { + const result = await this._pool.query( + `SELECT count(*) FROM location_events_1h WHERE room_id = $1 AND bucket_start >= $2 AND bucket_start <= $3`, + [roomId, new Date(fromMs), new Date(toMs)] + ); + if (Number(result.rows[0].count) <= maxPoints) return "1h"; + } catch { + // 1h table may not have data + } + } + + return "1d"; + } + + /** + * Query raw events for a room. + * @private + */ + async _queryRoomRaw(roomId, from, to, limit) { const params = [roomId]; const conditions = ["room_id = $1"]; let idx = 2; @@ -322,6 +1046,135 @@ export class PostgresAdapter { return result.rows; } + /** + * Query 1-minute downsampled data for a room. + * @private + */ + async _queryRoom1m(roomId, from, to, limit) { + const params = [roomId]; + const conditions = ["room_id = $1"]; + let idx = 2; + + if (from instanceof Date) { + conditions.push(`bucket_start >= $${idx++}`); + params.push(from.toISOString()); + } + if (to instanceof Date) { + conditions.push(`bucket_start <= $${idx++}`); + params.push(to.toISOString()); + } + + let sql = ` + SELECT + gen_random_uuid() AS id, + 'downsample' AS "clientId", + room_id AS "roomId", + latitude, longitude, + altitude, + NULL AS accuracy, + max_speed AS speed, + bucket_start::text AS timestamp, + created_at::text AS "createdAt" + FROM location_events_1m + WHERE ${conditions.join(" AND ")} + ORDER BY bucket_start ASC + `; + + if (typeof limit === "number" && limit > 0) { + sql += ` LIMIT $${idx}`; + params.push(limit); + } + + const result = await this._pool.query(sql, params); + return result.rows; + } + + /** + * Query 1-hour aggregate data for a room. + * @private + */ + async _queryRoom1h(roomId, from, to, limit) { + const params = [roomId]; + const conditions = ["room_id = $1"]; + let idx = 2; + + if (from instanceof Date) { + conditions.push(`bucket_start >= $${idx++}`); + params.push(from.toISOString()); + } + if (to instanceof Date) { + conditions.push(`bucket_start <= $${idx++}`); + params.push(to.toISOString()); + } + + let sql = ` + SELECT + gen_random_uuid() AS id, + 'aggregate-1h' AS "clientId", + room_id AS "roomId", + latitude, longitude, + altitude, + NULL AS accuracy, + avg_speed AS speed, + bucket_start::text AS timestamp, + created_at::text AS "createdAt" + FROM location_events_1h + WHERE ${conditions.join(" AND ")} + ORDER BY bucket_start ASC + `; + + if (typeof limit === "number" && limit > 0) { + sql += ` LIMIT $${idx}`; + params.push(limit); + } + + const result = await this._pool.query(sql, params); + return result.rows; + } + + /** + * Query 1-day aggregate data for a room. + * @private + */ + async _queryRoom1d(roomId, from, to, limit) { + const params = [roomId]; + const conditions = ["room_id = $1"]; + let idx = 2; + + if (from instanceof Date) { + conditions.push(`bucket_start >= $${idx++}`); + params.push(from.toISOString()); + } + if (to instanceof Date) { + conditions.push(`bucket_start <= $${idx++}`); + params.push(to.toISOString()); + } + + let sql = ` + SELECT + gen_random_uuid() AS id, + 'aggregate-1d' AS "clientId", + room_id AS "roomId", + latitude, longitude, + altitude, + NULL AS accuracy, + avg_speed AS speed, + bucket_start::text AS timestamp, + created_at::text AS "createdAt" + FROM location_events_1d + WHERE ${conditions.join(" AND ")} + ORDER BY bucket_start ASC + `; + + if (typeof limit === "number" && limit > 0) { + sql += ` LIMIT $${idx}`; + params.push(limit); + } + + const result = await this._pool.query(sql, params); + return result.rows; + } + /** * Returns events whose coordinates fall within the bounding box. * Uses the composite (latitude, longitude) index for efficient range scans. @@ -387,8 +1240,9 @@ export class PostgresAdapter { // ─────────────────────────── shutdown ───────────────────────────────────── /** - * Flushes any remaining buffered events, stops the flush timer, and drains - * the connection pool. Idempotent. + * Flushes any remaining buffered events, stops the flush timer, + * stops the compaction timer, and drains the connection pool. + * Idempotent. * * @returns {Promise} */ @@ -396,15 +1250,21 @@ export class PostgresAdapter { if (this._closed) return; this._closed = true; - // Stop the periodic timer. + // Stop the periodic flush timer. if (this._flushTimer) { clearInterval(this._flushTimer); this._flushTimer = null; } + // Stop the compaction timer. + if (this._compactionTimer) { + clearTimeout(this._compactionTimer); + this._compactionTimer = null; + } + // Attempt a final flush of any remaining events. if (this._buffer.length > 0 && this._pool) { - this._flushing = false; // reset flag so flush can proceed + this._flushing = false; try { await this._flush(); } catch (err) { diff --git a/tests/storage-compaction.test.js b/tests/storage-compaction.test.js new file mode 100644 index 0000000..fbb5fee --- /dev/null +++ b/tests/storage-compaction.test.js @@ -0,0 +1,617 @@ +/** + * @fileoverview Compaction and resolution-aware query tests for StorageAdapter. + * + * Tests the MemoryAdapter implementation of compaction, downsampling, + * resolution-aware queryRoom, and getCompactionStatus. + * PostgreSQL integration tests are skipped when DATABASE_URL is not set. + */ + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { MemoryAdapter } from "../src/storage/memory.js"; +import { assertStorageAdapter } from "../src/storage/adapter.js"; + +// ─── helpers ────────────────────────────────────────────────────────────────── + +/** + * Returns a valid LocationEvent with all required fields. + * @param {Partial} [overrides] + * @returns {import("../src/storage/adapter.js").LocationEvent} + */ +function makeEvent(overrides = {}) { + return { + clientId: "client-001", + roomId: "room-alpha", + latitude: 40.7128, + longitude: -74.006, + altitude: 10, + accuracy: 5, + speed: 1.2, + timestamp: new Date().toISOString(), + ...overrides, + }; +} + +/** + * Generate a series of events spread across a time range. + * @param {string} roomId + * @param {number} startMs - Start timestamp in ms. + * @param {number} count - Number of events. + * @param {number} intervalMs - Interval between events. + * @returns {import("../src/storage/adapter.js").LocationEvent[]} + */ +function generateEvents(roomId, startMs, count, intervalMs = 1000) { + const events = []; + for (let i = 0; i < count; i++) { + events.push(makeEvent({ + roomId, + clientId: `client-${String(i).padStart(3, "0")}`, + latitude: 40.7128 + (i * 0.001), + longitude: -74.006 + (i * 0.001), + altitude: 10 + (i * 0.1), + speed: 1.0 + (i * 0.01), + timestamp: new Date(startMs + i * intervalMs).toISOString(), + })); + } + return events; +} + +// ─── adapter interface contract ─────────────────────────────────────────────── + +describe("StorageAdapter interface with compaction", () => { + let adapter; + + beforeEach(() => { + adapter = new MemoryAdapter(); + }); + + afterEach(async () => { + await adapter.close(); + }); + + it("implements the full StorageAdapter interface including compact and getCompactionStatus", () => { + expect(() => assertStorageAdapter(adapter)).not.toThrow(); + expect(typeof adapter.compact).toBe("function"); + expect(typeof adapter.getCompactionStatus).toBe("function"); + }); + + it("getCompactionStatus returns valid structure", async () => { + const status = await adapter.getCompactionStatus(); + expect(status).toHaveProperty("lastRun"); + expect(status).toHaveProperty("nextRun"); + expect(status).toHaveProperty("tiers"); + expect(Array.isArray(status.tiers)).toBe(true); + expect(status.tiers.length).toBe(4); + + const tierNames = status.tiers.map((t) => t.name); + expect(tierNames).toContain("raw"); + expect(tierNames).toContain("1m"); + expect(tierNames).toContain("1h"); + expect(tierNames).toContain("1d"); + + for (const tier of status.tiers) { + expect(typeof tier.rows).toBe("number"); + expect(typeof tier.sizeBytes).toBe("number"); + } + }); +}); + +// ─── queryRoom resolution routing ───────────────────────────────────────────── + +describe("queryRoom resolution routing", () => { + let adapter; + + beforeEach(async () => { + adapter = new MemoryAdapter(); + }); + + afterEach(async () => { + await adapter.close(); + }); + + it("default resolution is 'auto'", async () => { + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date().toISOString() }), + ]); + // Should not throw with default resolution + const results = await adapter.queryRoom("r1"); + expect(results.length).toBeGreaterThan(0); + }); + + it("resolution='raw' returns raw events", async () => { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date(base).toISOString() }), + makeEvent({ roomId: "r1", timestamp: new Date(base + 1000).toISOString() }), + ]); + + const results = await adapter.queryRoom("r1", { resolution: "raw" }); + expect(results).toHaveLength(2); + expect(results[0].clientId).not.toBe("downsample"); + }); + + it("resolution='1m' returns downsampled data after compaction", async () => { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date(base).toISOString() }), + makeEvent({ roomId: "r1", timestamp: new Date(base + 5000).toISOString() }), + makeEvent({ roomId: "r1", timestamp: new Date(base + 10000).toISOString() }), + ]); + + await adapter.compact(); + + const results = await adapter.queryRoom("r1", { resolution: "1m" }); + expect(results.length).toBeGreaterThan(0); + // Downsampled events have clientId "downsample" + expect(results[0].clientId).toBe("downsample"); + }); + + it("resolution='1h' returns aggregate data after compaction", async () => { + const base = Date.now(); + // Generate events across multiple minutes + const events = []; + for (let i = 0; i < 10; i++) { + events.push(makeEvent({ + roomId: "r1", + timestamp: new Date(base + i * 60000).toISOString(), + })); + } + await adapter.writeBatch(events); + await adapter.compact(); + + const results = await adapter.queryRoom("r1", { + resolution: "1h", + from: new Date(base - 60000), + to: new Date(base + 600000), + }); + expect(results.length).toBeGreaterThan(0); + expect(results[0].clientId).toBe("aggregate-1h"); + }); + + it("resolution='1d' returns aggregate data after compaction", async () => { + const base = Date.now(); + // Generate events across multiple hours + const events = []; + for (let i = 0; i < 5; i++) { + events.push(makeEvent({ + roomId: "r1", + timestamp: new Date(base + i * 3600000).toISOString(), + })); + } + await adapter.writeBatch(events); + await adapter.compact(); + + const results = await adapter.queryRoom("r1", { + resolution: "1d", + from: new Date(base - 3600000), + to: new Date(base + 5 * 3600000), + }); + expect(results.length).toBeGreaterThan(0); + expect(results[0].clientId).toBe("aggregate-1d"); + }); + + it("resolution='auto' selects raw for short time ranges", async () => { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date(base).toISOString() }), + ]); + + const results = await adapter.queryRoom("r1", { + from: new Date(base - 10000), + to: new Date(base + 10000), + resolution: "auto", + }); + // Short range should return raw events + expect(results.length).toBe(1); + }); + + it("resolution='auto' defaults to raw when no time range specified", async () => { + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date().toISOString() }), + ]); + + const results = await adapter.queryRoom("r1", { resolution: "auto" }); + expect(results.length).toBeGreaterThan(0); + expect(results[0].clientId).not.toBe("downsample"); + }); +}); + +// ─── compaction pipeline ────────────────────────────────────────────────────── + +describe("MemoryAdapter compaction", () => { + let adapter; + + beforeEach(async () => { + adapter = new MemoryAdapter(); + }); + + afterEach(async () => { + await adapter.close(); + }); + + it("compact() returns CompactionResult with correct shape", async () => { + await adapter.writeBatch([ + makeEvent({ roomId: "r1" }), + ]); + + const result = await adapter.compact(); + expect(result).toHaveProperty("rawDeleted"); + expect(result).toHaveProperty("downsample1m"); + expect(result).toHaveProperty("aggregate1h"); + expect(result).toHaveProperty("aggregate1d"); + expect(result).toHaveProperty("durationMs"); + expect(typeof result.rawDeleted).toBe("number"); + expect(typeof result.downsample1m).toBe("number"); + expect(typeof result.durationMs).toBe("number"); + }); + + it("compact() computes 1m downsamples from raw data", async () => { + const base = Date.now(); + // Generate multiple events within the same minute + const events = []; + for (let i = 0; i < 5; i++) { + events.push(makeEvent({ + roomId: "r1", + latitude: 40.7128 + i * 0.01, + longitude: -74.006 + i * 0.01, + timestamp: new Date(base + i * 1000).toISOString(), + })); + } + await adapter.writeBatch(events); + + const result = await adapter.compact(); + // Should have created downsample rows + expect(result.downsample1m).toBeGreaterThanOrEqual(1); + + // Verify 1m data is queryable + const m1Results = await adapter.queryRoom("r1", { resolution: "1m" }); + expect(m1Results.length).toBeGreaterThanOrEqual(1); + }); + + it("compact() computes 1h aggregates from 1m data", async () => { + const base = Date.now(); + // Generate events across multiple minutes to get 1h aggregates + const events = []; + for (let i = 0; i < 10; i++) { + events.push(makeEvent({ + roomId: "r1", + timestamp: new Date(base + i * 60000).toISOString(), + })); + } + await adapter.writeBatch(events); + + const result = await adapter.compact(); + expect(result.aggregate1h).toBeGreaterThanOrEqual(1); + }); + + it("compact() computes 1d aggregates from 1h data", async () => { + const base = Date.now(); + // Generate events across multiple hours + const events = []; + for (let i = 0; i < 5; i++) { + events.push(makeEvent({ + roomId: "r1", + timestamp: new Date(base + i * 3600000).toISOString(), + })); + } + await adapter.writeBatch(events); + + const result = await adapter.compact(); + expect(result.aggregate1d).toBeGreaterThanOrEqual(1); + }); + + it("compact() with custom retention overrides", async () => { + await adapter.writeBatch([ + makeEvent({ roomId: "r1" }), + ]); + + const result = await adapter.compact({ + rawRetentionDays: 1, + downsample1mRetentionDays: 30, + downsample1hRetentionDays: 180, + aggregate1dRetentionDays: 1000, + }); + expect(result.rawDeleted).toBeGreaterThanOrEqual(0); + }); + + it("compact() is idempotent — running twice produces same results", async () => { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date(base).toISOString() }), + makeEvent({ roomId: "r1", timestamp: new Date(base + 5000).toISOString() }), + ]); + + await adapter.compact(); + const result2 = await adapter.compact(); + + // Second run should still work + expect(result2.durationMs).toBeGreaterThanOrEqual(0); + }); + + it("getCompactionStatus updates after compact()", async () => { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "r1", timestamp: new Date(base).toISOString() }), + ]); + + const statusBefore = await adapter.getCompactionStatus(); + expect(statusBefore.lastRun).toBeNull(); + + await adapter.compact(); + + const statusAfter = await adapter.getCompactionStatus(); + expect(statusAfter.lastRun).not.toBeNull(); + expect(statusAfter.tiers[0].rows).toBeGreaterThanOrEqual(0); + }); + + it("compaction is idempotent and resumable", async () => { + const base = Date.now(); + await adapter.writeBatch(generateEvents("r1", base, 50, 1000)); + await adapter.compact(); + + // Run compaction again — should not throw or duplicate data + const result = await adapter.compact(); + expect(result.durationMs).toBeGreaterThanOrEqual(0); + + const m1 = await adapter.queryRoom("r1", { resolution: "1m" }); + expect(m1.length).toBeGreaterThanOrEqual(1); + }); +}); + +// ─── data tier statistics ───────────────────────────────────────────────────── + +describe("Compaction tier statistics", () => { + let adapter; + + beforeEach(async () => { + adapter = new MemoryAdapter(); + }); + + afterEach(async () => { + await adapter.close(); + }); + + it("raw tier shows correct row count", async () => { + await adapter.writeBatch(generateEvents("r1", Date.now(), 25)); + const status = await adapter.getCompactionStatus(); + const rawTier = status.tiers.find((t) => t.name === "raw"); + expect(rawTier.rows).toBe(25); + }); + + it("raw tier oldest/newest timestamps are correct", async () => { + const base = Date.now(); + await adapter.writeBatch(generateEvents("r1", base, 5, 60000)); + const status = await adapter.getCompactionStatus(); + const rawTier = status.tiers.find((t) => t.name === "raw"); + expect(rawTier.oldest).not.toBeNull(); + expect(rawTier.newest).not.toBeNull(); + }); + + it("1m tier gets populated after compaction", async () => { + const base = Date.now(); + await adapter.writeBatch(generateEvents("r1", base, 10, 1000)); + await adapter.compact(); + + const status = await adapter.getCompactionStatus(); + const m1Tier = status.tiers.find((t) => t.name === "1m"); + expect(m1Tier.rows).toBeGreaterThanOrEqual(1); + }); + + it("1h tier gets populated after compaction", async () => { + const base = Date.now(); + // Events across multiple minutes + await adapter.writeBatch(generateEvents("r1", base, 10, 60000)); + await adapter.compact(); + + const status = await adapter.getCompactionStatus(); + const h1Tier = status.tiers.find((t) => t.name === "1h"); + expect(h1Tier.rows).toBeGreaterThanOrEqual(1); + }); + + it("sizeBytes are non-negative", async () => { + await adapter.writeBatch(generateEvents("r1", Date.now(), 10)); + await adapter.compact(); + const status = await adapter.getCompactionStatus(); + for (const tier of status.tiers) { + expect(tier.sizeBytes).toBeGreaterThanOrEqual(0); + } + }); +}); + +// ─── multi-room compaction ──────────────────────────────────────────────────── + +describe("Multi-room compaction", () => { + let adapter; + + beforeEach(async () => { + adapter = new MemoryAdapter(); + }); + + afterEach(async () => { + await adapter.close(); + }); + + it("compaction computes per-room downsamples", async () => { + const base = Date.now(); + await adapter.writeBatch([ + ...generateEvents("room-a", base, 5, 1000), + ...generateEvents("room-b", base, 5, 1000), + ]); + + await adapter.compact(); + + const a1m = await adapter.queryRoom("room-a", { resolution: "1m" }); + const b1m = await adapter.queryRoom("room-b", { resolution: "1m" }); + expect(a1m.length).toBeGreaterThanOrEqual(1); + expect(b1m.length).toBeGreaterThanOrEqual(1); + }); + + it("queryRoom resolution only returns data for the requested room", async () => { + const base = Date.now(); + await adapter.writeBatch([ + ...generateEvents("room-a", base, 5, 1000), + ...generateEvents("room-b", base, 5, 1000), + ]); + + await adapter.compact(); + + const aResults = await adapter.queryRoom("room-a", { resolution: "1m" }); + for (const r of aResults) { + expect(r.roomId).toBe("room-a"); + } + }); +}); + +// ─── haversine distance calculation ─────────────────────────────────────────── + +describe("Compaction distance calculation", () => { + let adapter; + + beforeEach(async () => { + adapter = new MemoryAdapter(); + }); + + afterEach(async () => { + await adapter.close(); + }); + + it("1h aggregate includes totalDistance", async () => { + const base = Date.now(); + // Events spread across different locations over multiple minutes + const events = []; + for (let i = 0; i < 10; i++) { + events.push(makeEvent({ + roomId: "r1", + latitude: 40.7128 + i * 0.001, + longitude: -74.006 + i * 0.001, + timestamp: new Date(base + i * 60000).toISOString(), + })); + } + await adapter.writeBatch(events); + await adapter.compact(); + + const h1 = await adapter.queryRoom("r1", { + resolution: "1h", + from: new Date(base - 60000), + to: new Date(base + 600000), + }); + expect(h1.length).toBeGreaterThanOrEqual(1); + }); +}); + +// ─── close edge cases ───────────────────────────────────────────────────────── + +describe("Compaction close edge cases", () => { + it("compact() throws after close()", async () => { + const adapter = new MemoryAdapter(); + await adapter.close(); + await expect(adapter.compact()).rejects.toThrow("MemoryAdapter is closed"); + }); + + it("getCompactionStatus() throws after close()", async () => { + const adapter = new MemoryAdapter(); + await adapter.close(); + await expect(adapter.getCompactionStatus()).rejects.toThrow("MemoryAdapter is closed"); + }); + + it("queryRoom with resolution throws after close()", async () => { + const adapter = new MemoryAdapter(); + await adapter.close(); + await expect( + adapter.queryRoom("r1", { resolution: "1m" }) + ).rejects.toThrow("MemoryAdapter is closed"); + }); +}); + +// ─── PostgreSQL integration tests (skipped without DATABASE_URL) ───────────── + +const describePg = process.env.DATABASE_URL ? describe : describe.skip; + +describePg("PostgreSQL compaction integration", () => { + let PostgresAdapter; + + beforeEach(async () => { + const mod = await import("../src/storage/postgres.js"); + PostgresAdapter = mod.PostgresAdapter; + }); + + it("PostgresAdapter implements compact and getCompactionStatus", async () => { + const adapter = new PostgresAdapter({ + connectionString: process.env.DATABASE_URL, + compactionIntervalMs: 600000, // disable auto-compaction for tests + }); + try { + expect(() => assertStorageAdapter(adapter)).not.toThrow(); + expect(typeof adapter.compact).toBe("function"); + expect(typeof adapter.getCompactionStatus).toBe("function"); + + const status = await adapter.getCompactionStatus(); + expect(status).toHaveProperty("tiers"); + expect(Array.isArray(status.tiers)).toBe(true); + } finally { + await adapter.close(); + } + }); + + it("PostgresAdapter compact() runs without error", async () => { + const adapter = new PostgresAdapter({ + connectionString: process.env.DATABASE_URL, + compactionIntervalMs: 600000, + }); + try { + const result = await adapter.compact(); + expect(result).toHaveProperty("rawDeleted"); + expect(result).toHaveProperty("downsample1m"); + expect(result).toHaveProperty("aggregate1h"); + expect(result).toHaveProperty("aggregate1d"); + expect(result).toHaveProperty("durationMs"); + } finally { + await adapter.close(); + } + }); + + it("PostgresAdapter queryRoom with resolution=raw returns raw events", async () => { + const adapter = new PostgresAdapter({ + connectionString: process.env.DATABASE_URL, + compactionIntervalMs: 600000, + }); + try { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "pg-test-raw", timestamp: new Date(base).toISOString() }), + makeEvent({ roomId: "pg-test-raw", timestamp: new Date(base + 1000).toISOString() }), + ]); + + // Flush the buffer + await adapter._flush(); + + const results = await adapter.queryRoom("pg-test-raw", { resolution: "raw" }); + expect(results.length).toBeGreaterThanOrEqual(2); + } finally { + await adapter.close(); + } + }); + + it("PostgresAdapter queryRoom with resolution=auto selects appropriate tier", async () => { + const adapter = new PostgresAdapter({ + connectionString: process.env.DATABASE_URL, + compactionIntervalMs: 600000, + }); + try { + const base = Date.now(); + await adapter.writeBatch([ + makeEvent({ roomId: "pg-test-auto", timestamp: new Date(base).toISOString() }), + ]); + await adapter._flush(); + + // Short range should resolve to raw + const results = await adapter.queryRoom("pg-test-auto", { + from: new Date(base - 10000), + to: new Date(base + 10000), + resolution: "auto", + }); + expect(results.length).toBeGreaterThanOrEqual(1); + } finally { + await adapter.close(); + } + }); +});