Develop update

.github/workflows/ci.yml

@@ -75,6 +75,9 @@ jobs:
       - name: Build shared package
         run: pnpm --filter '@logtide/shared' build
 
+      - name: Build reservoir package
+        run: pnpm --filter '@logtide/reservoir' build
+
       - name: Run backend tests with coverage
         working-directory: packages/backend
         env:
@@ -117,6 +120,55 @@ jobs:
           fi
           echo "::notice::Coverage $COVERAGE% meets the 80% threshold"
 
+  # ====================
+  # Reservoir Tests
+  # ====================
+  reservoir-test:
+    name: Reservoir Tests
+    runs-on: ubuntu-latest
+
+    services:
+      clickhouse:
+        image: clickhouse/clickhouse-server:24.1
+        env:
+          CLICKHOUSE_DB: logtide_test
+          CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
+        ports:
+          - 18123:8123
+        options: >-
+          --health-cmd "clickhouse-client --query 'SELECT 1'"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: ${{ env.PNPM_VERSION }}
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: ${{ env.NODE_VERSION }}
+          cache: 'pnpm'
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Build shared package
+        run: pnpm --filter '@logtide/shared' build
+
+      - name: Build reservoir package
+        run: pnpm --filter '@logtide/reservoir' build
+
+      - name: Run reservoir tests
+        working-directory: packages/reservoir
+        run: npx vitest run --reporter=verbose
+
   # ====================
   # Typecheck
   # ====================
@@ -145,6 +197,9 @@ jobs:
       - name: Build shared package
         run: pnpm --filter '@logtide/shared' build
 
+      - name: Build reservoir package
+        run: pnpm --filter '@logtide/reservoir' build
+
       - name: Typecheck backend
         run: pnpm --filter '@logtide/backend' typecheck
 
@@ -227,7 +282,7 @@ jobs:
   build:
     name: Build Docker Images
     runs-on: ubuntu-latest
-    needs: [backend-test, typecheck, e2e-test]
+    needs: [backend-test, reservoir-test, typecheck, e2e-test]
 
     steps:
       - name: Checkout

packages/backend/Dockerfile

@@ -9,13 +9,15 @@ WORKDIR /app
 # Copy workspace files
 COPY pnpm-workspace.yaml package.json pnpm-lock.yaml* tsconfig.base.json ./
 COPY packages/shared ./packages/shared
+COPY packages/reservoir ./packages/reservoir
 COPY packages/backend ./packages/backend
 
 # Install dependencies
 RUN pnpm install --frozen-lockfile
 
 # Build packages in order
 RUN pnpm --filter '@logtide/shared' build
+RUN pnpm --filter '@logtide/reservoir' build
 RUN pnpm --filter '@logtide/backend' build
 
 # Production stage
@@ -29,13 +31,15 @@ WORKDIR /app
 # Copy workspace files
 COPY pnpm-workspace.yaml package.json pnpm-lock.yaml ./
 COPY packages/shared/package.json ./packages/shared/
+COPY packages/reservoir/package.json ./packages/reservoir/
 COPY packages/backend/package.json ./packages/backend/
 
 # Install production dependencies only
 RUN pnpm install --frozen-lockfile --prod
 
 # Copy built files
 COPY --from=builder /app/packages/shared/dist ./packages/shared/dist
+COPY --from=builder /app/packages/reservoir/dist ./packages/reservoir/dist
 COPY --from=builder /app/packages/backend/dist ./packages/backend/dist
 COPY --from=builder /app/packages/backend/migrations ./packages/backend/migrations
 

packages/backend/migrations/023_optimize_slow_queries.sql

@@ -14,6 +14,10 @@
 -- The most effective optimization for hostname queries is keeping the time
 -- window short (6h default = ~350ms vs 7d = ~3s+).
 
+-- Disable statement timeout for this migration.
+-- ANALYZE on large hypertables can take minutes and will exceed the default pool timeout.
+SET statement_timeout = '0';
+
 -- Reduce chunk interval from 7 days to 1 day.
 -- With ~4.5 GB/day ingestion, weekly chunks grow to 31 GB uncompressed
 -- before compression can kick in (only works on closed chunks).
@@ -23,3 +27,6 @@ SELECT set_chunk_time_interval('logs', INTERVAL '1 day');
 
 -- Run ANALYZE to ensure planner has up-to-date statistics.
 ANALYZE logs;
+
+-- Reset statement timeout to default (pool-level setting will apply on next query).
+RESET statement_timeout;

packages/backend/package.json

@@ -23,7 +23,7 @@
     "test:coverage": "node src/scripts/run-tests.mjs --coverage",
     "test:ci": "vitest run",
     "test:ci:coverage": "vitest run --coverage",
-    "typecheck": "tsc --noEmit",
+    "typecheck": "pnpm --filter @logtide/shared --filter @logtide/reservoir build && tsc --noEmit",
     "clean": "rm -rf dist",
     "load:smoke": "k6 run load-tests/smoke.js",
     "load:ingestion": "k6 run load-tests/ingestion.js",
@@ -33,7 +33,8 @@
     "load:e2e:query": "bash scripts/run-load-tests.sh query",
     "load:e2e:all": "bash scripts/run-load-tests.sh all",
     "seed:load-test": "tsx src/scripts/seed-load-test.ts",
-    "seed:massive": "tsx src/scripts/seed-massive-data.ts"
+    "seed:massive": "tsx src/scripts/seed-massive-data.ts",
+    "migrate:storage": "tsx src/scripts/migrate-storage.ts"
   },
   "dependencies": {
     "@fastify/cors": "^10.0.2",

packages/backend/src/database/reservoir.ts

@@ -25,6 +25,7 @@ function createReservoir(): Reservoir {
       pool,
       tableName: 'logs',
       skipInitialize: true,
+      projectIdType: 'uuid',
     },
   );
 }

packages/backend/src/modules/admin/service.ts

@@ -482,24 +482,24 @@ export class AdminService {
             ),
 
             reservoir.count({ from: oneHourAgo, to: now })
-                .then(r => ({ count: r.count })),
+                .then((r: { count: number }) => ({ count: r.count })),
 
             reservoir.count({ from: oneDayAgo, to: now })
-                .then(r => ({ count: r.count })),
+                .then((r: { count: number }) => ({ count: r.count })),
         ]);
 
         return {
             total: totalLogs.count,
-            perDay: logsPerDay.rows.map((row) => ({
+            perDay: logsPerDay.rows.map((row: { date: string; count: number }) => ({
                 date: row.date,
                 count: row.count,
             })),
-            topOrganizations: topOrgs.rows.map((org) => ({
+            topOrganizations: topOrgs.rows.map((org: { organizationId: string; organizationName: string; count: number }) => ({
                 organizationId: org.organizationId,
                 organizationName: org.organizationName,
                 count: org.count,
             })),
-            topProjects: topProjects.rows.map((proj) => ({
+            topProjects: topProjects.rows.map((proj: { projectId: string; projectName: string; organizationName: string; count: number }) => ({
                 projectId: proj.projectId,
                 projectName: proj.projectName,
                 organizationName: proj.organizationName,
@@ -553,11 +553,11 @@ export class AdminService {
 
         // Per-day counts from aggregate
         const perDay = logsPerDayResult.timeseries
-            .map(b => ({
+            .map((b: { bucket: Date; total: number }) => ({
                 date: b.bucket.toISOString().split('T')[0],
                 count: b.total,
             }))
-            .sort((a, b) => b.date.localeCompare(a.date))
+            .sort((a: { date: string }, b: { date: string }) => b.date.localeCompare(a.date))
             .slice(0, 30);
 
         // Top orgs/projects: count per project via reservoir, then aggregate by org
@@ -1781,7 +1781,7 @@ export class AdminService {
                     interval: '1h',
                 });
                 logsTimeline = {
-                    rows: aggResult.timeseries.map(b => ({
+                    rows: aggResult.timeseries.map((b: { bucket: Date; total: number }) => ({
                         bucket: b.bucket.toISOString(),
                         count: b.total,
                     })),

packages/backend/src/modules/alerts/baseline-calculator.ts

@@ -240,15 +240,15 @@ export class BaselineCalculatorService {
       if (aggResult.timeseries.length === 0) return null;
 
       // Sum only the requested levels per bucket
-      counts = aggResult.timeseries.map((bucket) => {
+      counts = aggResult.timeseries.map((bucket: { byLevel?: Record<string, number> }) => {
         let count = 0;
         if (bucket.byLevel) {
           for (const level of levels) {
             count += bucket.byLevel[level as ReservoirLogLevel] || 0;
           }
         }
         return count;
-      }).filter(c => c > 0).sort((a, b) => a - b);
+      }).filter((c: number) => c > 0).sort((a: number, b: number) => a - b);
 
       if (counts.length === 0) return null;
     }

packages/backend/src/modules/alerts/service.ts

@@ -774,7 +774,7 @@ export class AlertsService {
     for (const bucket of aggResult.timeseries) {
       let count = 0;
       if (bucket.byLevel) {
-        for (const [lvl, n] of Object.entries(bucket.byLevel)) {
+        for (const [lvl, n] of Object.entries(bucket.byLevel) as [string, number][]) {
           if (levelSet.has(lvl as LogLevel)) count += n;
         }
       }
@@ -873,7 +873,7 @@ export class AlertsService {
       sortOrder: 'desc',
     });
 
-    return result.logs.map((log) => ({
+    return result.logs.map((log: { time: Date; service: string; level: string; message: string; traceId?: string }) => ({
       time: log.time,
       service: log.service,
       level: log.level,
@@ -903,7 +903,7 @@ export class AlertsService {
       service: svcFilter,
     });
 
-    return result.values.filter((s) => s !== 'unknown');
+    return result.values.filter((s: string) => s !== 'unknown');
   }
 
   /**

packages/backend/src/modules/correlation/service.ts

@@ -261,14 +261,14 @@ export class CorrelationService {
     // Fetch actual logs (reservoir: works with any engine)
     const logsResult = await reservoir.getByIds({ ids: logIds, projectId });
     // Sort chronologically (getByIds returns in unspecified order)
-    const sortedLogs = logsResult.sort((a, b) => a.time.getTime() - b.time.getTime());
+    const sortedLogs = logsResult.sort((a: { time: Date }, b: { time: Date }) => a.time.getTime() - b.time.getTime());
 
     return {
       identifier: {
         type: identifierType,
         value: identifierValue,
       },
-      logs: sortedLogs.map((log) => ({
+      logs: sortedLogs.map((log: { id: string; time: Date; service: string; level: string; message: string; metadata?: Record<string, unknown>; traceId?: string; projectId: string }) => ({
         id: log.id,
         time: log.time,
         service: log.service,

packages/backend/src/modules/dashboard/service.ts

@@ -334,10 +334,10 @@ class DashboardService {
         to: lastHour,
         interval: '1h',
       });
-      historicalResults = aggResult.timeseries.flatMap((bucket) => {
+      historicalResults = aggResult.timeseries.flatMap((bucket: { bucket: Date; byLevel?: Record<string, number> }) => {
         const entries: Array<{ bucket: string | Date; level: string; count: string }> = [];
         if (bucket.byLevel) {
-          for (const [level, count] of Object.entries(bucket.byLevel)) {
+          for (const [level, count] of Object.entries(bucket.byLevel) as [string, number][]) {
             if (count > 0) {
               entries.push({ bucket: bucket.bucket, level, count: String(count) });
             }
@@ -354,10 +354,10 @@ class DashboardService {
       to: now,
       interval: '1h',
     });
-    recentResults = recentAgg.timeseries.flatMap((bucket) => {
+    recentResults = recentAgg.timeseries.flatMap((bucket: { bucket: Date; byLevel?: Record<string, number> }) => {
       const entries: Array<{ bucket: string; level: string; count: string }> = [];
       if (bucket.byLevel) {
-        for (const [level, count] of Object.entries(bucket.byLevel)) {
+        for (const [level, count] of Object.entries(bucket.byLevel) as [string, number][]) {
           if (count > 0) {
             entries.push({ bucket: bucket.bucket.toISOString(), level, count: String(count) });
           }
@@ -540,7 +540,7 @@ class DashboardService {
     const total = totalResult.count;
     if (total === 0) return [];
 
-    return topResult.values.map((s) => ({
+    return topResult.values.map((s: { value: string; count: number }) => ({
       name: s.value,
       count: s.count,
       percentage: Math.round((s.count / total) * 100),
@@ -718,7 +718,7 @@ class DashboardService {
       sortOrder: 'desc',
     });
 
-    const result = queryResult.logs.map((e) => ({
+    const result = queryResult.logs.map((e: { time: Date; service: string; level: string; message: string; projectId: string; traceId?: string }) => ({
       time: e.time.toISOString(),
       service: e.service,
       level: e.level as 'error' | 'critical',

packages/backend/src/modules/exceptions/service.ts

@@ -474,15 +474,15 @@ export class ExceptionService {
     const storedLogs = await reservoir.getByIds({ ids: logIds, projectId });
 
     // Build lookup map and return in the same order
-    const logMap = new Map(storedLogs.map(l => [l.id, l]));
+    const logMap = new Map(storedLogs.map((l: { id: string; time: Date; service: string; message: string }) => [l.id, l]));
     const logs = logIds
       .map(id => logMap.get(id))
-      .filter(Boolean)
+      .filter((l): l is { id: string; time: Date; service: string; message: string } => Boolean(l))
       .map(l => ({
-        id: l!.id,
-        time: l!.time,
-        service: l!.service,
-        message: l!.message,
+        id: l.id,
+        time: l.time,
+        service: l.service,
+        message: l.message,
       }));
 
     return {

packages/backend/src/modules/ingestion/service.ts

@@ -90,7 +90,7 @@ export class IngestionService {
 
     // Insert via reservoir (raw parametrized SQL with RETURNING *)
     const ingestResult = await reservoir.ingestReturning(records);
-    const insertedLogs = ingestResult.rows.map((row) => ({
+    const insertedLogs = ingestResult.rows.map((row: { id: string; time: Date; projectId: string; service: string; level: string; message: string; metadata?: Record<string, unknown>; traceId?: string; spanId?: string }) => ({
       id: row.id,
       time: row.time,
       project_id: row.projectId,

packages/backend/src/modules/query/service.ts

@@ -166,7 +166,7 @@ export class QueryService {
       limit: 1000,
     });
 
-    const result = queryResult.logs.map(log => ({
+    const result = queryResult.logs.map((log: { id: string; time: Date; projectId: string; service: string; level: string; message: string; metadata?: Record<string, unknown>; traceId?: string }) => ({
       id: log.id,
       time: log.time,
       projectId: log.projectId,
@@ -342,7 +342,7 @@ export class QueryService {
         to: to ?? new Date(),
         limit,
       });
-      result = topResult.values.map(v => ({ service: v.value, count: v.count }));
+      result = topResult.values.map((v: { value: string; count: number }) => ({ service: v.value, count: v.count }));
     }
 
     // Cache aggregation results
@@ -511,7 +511,7 @@ export class QueryService {
       level: ['error', 'critical'],
       limit,
     });
-    const result = topResult.values.map(v => ({ message: v.value, count: v.count }));
+    const result = topResult.values.map((v: { value: string; count: number }) => ({ message: v.value, count: v.count }));
 
     // Cache aggregation results
     await CacheManager.set(cacheKey, result, CACHE_TTL.STATS);