feat(storage): implement time-series compaction with downsampling and retention

.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

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

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);

src/rate-limiter.js

@@ -83,3 +83,5 @@ export function createRateLimiter(maxPerSecond) {
     get size() {
       return windows.size;
     },
+  };
+}

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 };
 }

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<LocationEvent[]>} 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<LocationEvent[]>} querySpatial
  *   Return events whose coordinates fall within the supplied bounding box.
@@ -71,6 +110,15 @@
  * @property {function(): Promise<void>} close
  *   Release all resources held by the adapter (connections, timers, …).
  *   Must be idempotent — calling it multiple times must not throw.
+ *
+ * @property {function(object=): Promise<CompactionResult>} 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<CompactionStatus>} 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(