feat(bff): add PgBouncer pooling, Redis cache with stampede protectio…

.coderabbit.yaml

@@ -20,3 +20,4 @@ reviews:
         Verify eventId and tenantId are always set.
 chat:
   auto_reply: true
+

apps/bff/src/datasources/cassandra.ts

@@ -69,3 +69,4 @@ export async function closeCassandraClient(): Promise<void> {
     client = null;
   }
 }
+

apps/bff/src/datasources/elasticsearch.ts

@@ -51,3 +51,4 @@ export const search = {
     }
   },
 };
+

apps/bff/src/datasources/postgres.ts

@@ -1,16 +1,16 @@
 import { Pool } from "pg";
 import { postgresCircuitBreaker } from "../lib/circuitBreaker";
+import { cache } from "./redis";
 
 const pool = new Pool({
-  host:     process.env.READ_DB_HOST     || "postgres-read",
+  host:     process.env.READ_DB_HOST     || "pgbouncer",
   port:     parseInt(process.env.READ_DB_PORT || "5432"),
   database: process.env.READ_DB_NAME     || "grainguard_read",
   user:     process.env.READ_DB_USER     || "postgres",
   password: process.env.READ_DB_PASSWORD || "postgres",
-  max: 10,
+  max: 50,
 });
 
-
 // Circuit-breaker-wrapped query helper
 async function cbQuery(text: string, values?: any[]): Promise<import("pg").QueryResult<any>> {
   return postgresCircuitBreaker.execute(() => pool.query(text, values));
@@ -33,9 +33,6 @@ export async function tenantQuery(
   });
 }
 
-// Tenant-scoped query — sets app.current_tenant_id for RLS enforcement
-// Use this for all queries that should be tenant-isolated
-
 export const db = {
 
   async getDevice(deviceId: string) {
@@ -49,14 +46,29 @@ export const db = {
   },
 
   async getAllDevices(limit: number = 20) {
-    const result = await cbQuery(
-      `SELECT device_id, tenant_id, serial_number, created_at
-       FROM device_projections
-       ORDER BY created_at DESC
-       LIMIT $1`,
-      [limit]
-    );
-    return result.rows;
+    const cacheKey = `devices:all:${limit}`;
+    const cached = await cache.get<any[]>(cacheKey);
+    if (cached) return cached;
+
+    const locked = await cache.acquireLock(cacheKey, 5);
+    if (!locked) {
+      await new Promise(r => setTimeout(r, 100));
+      return await cache.get<any[]>(cacheKey) || [];
+    }
+
+    try {
+      const result = await cbQuery(
+        `SELECT device_id, tenant_id, serial_number, created_at
+         FROM device_projections
+         ORDER BY created_at DESC
+         LIMIT $1`,
+        [limit]
+      );
+      await cache.set(cacheKey, result.rows, 30);
+      return result.rows;
+    } finally {
+      await cache.releaseLock(cacheKey);
+    }
   },
 
   async getDeviceTelemetry(deviceId: string) {
@@ -70,26 +82,43 @@ export const db = {
   },
 
   async getAllTelemetry(limit: number = 20, tenantId?: string) {
-    if (tenantId) {
+    const cacheKey = `telemetry:all:${tenantId || "global"}:${limit}`;
+    const cached = await cache.get<any[]>(cacheKey);
+    if (cached) return cached;
+
+    const locked = await cache.acquireLock(cacheKey, 5);
+    if (!locked) {
+      await new Promise(r => setTimeout(r, 100));
+      return await cache.get<any[]>(cacheKey) || [];
+    }
+
+    try {
+      if (tenantId) {
+        const result = await cbQuery(
+          `SELECT t.device_id, t.temperature, t.humidity, t.recorded_at, t.updated_at, t.version
+           FROM device_telemetry_latest t
+           INNER JOIN device_projections d ON d.device_id = t.device_id
+           WHERE d.tenant_id = $1
+           ORDER BY t.updated_at DESC
+           LIMIT $2`,
+          [tenantId, limit]
+        );
+        await cache.set(cacheKey, result.rows, 30);
+        return result.rows;
+      }
+
       const result = await cbQuery(
-        `SELECT t.device_id, t.temperature, t.humidity, t.recorded_at, t.updated_at, t.version
-         FROM device_telemetry_latest t
-         INNER JOIN device_projections d ON d.device_id = t.device_id
-         WHERE d.tenant_id = $1
-         ORDER BY t.updated_at DESC
-         LIMIT $2`,
-        [tenantId, limit]
+        `SELECT device_id, temperature, humidity, recorded_at, updated_at, version
+         FROM device_telemetry_latest
+         ORDER BY updated_at DESC
+         LIMIT $1`,
+        [limit]
       );
+      await cache.set(cacheKey, result.rows, 30);
       return result.rows;
+    } finally {
+      await cache.releaseLock(cacheKey);
     }
-    const result = await cbQuery(
-      `SELECT device_id, temperature, humidity, recorded_at, updated_at, version
-       FROM device_telemetry_latest
-       ORDER BY updated_at DESC
-       LIMIT $1`,
-      [limit]
-    );
-    return result.rows;
   },
 
   async getDeviceWithTelemetry(deviceId: string) {
@@ -133,45 +162,48 @@ export const db = {
   },
 
   async getAllDevicesWithTelemetry(limit: number = 20, tenantId?: string) {
-    if (tenantId) {
+    const cacheKey = `devices:telemetry:${tenantId || "global"}:${limit}`;
+    const cached = await cache.get<any[]>(cacheKey);
+    if (cached) return cached;
+
+    const locked = await cache.acquireLock(cacheKey, 5);
+    if (!locked) {
+      await new Promise(r => setTimeout(r, 100));
+      return await cache.get<any[]>(cacheKey) || [];
+    }
+
+    try {
+      if (tenantId) {
+        const result = await cbQuery(
+          `SELECT
+             d.device_id, d.tenant_id, d.serial_number, d.created_at,
+             t.temperature, t.humidity, t.recorded_at, t.version
+           FROM device_projections d
+           LEFT JOIN device_telemetry_latest t ON d.device_id = t.device_id
+           WHERE d.tenant_id = $1
+           ORDER BY d.created_at DESC
+           LIMIT $2`,
+          [tenantId, limit]
+        );
+        await cache.set(cacheKey, result.rows, 30);
+        return result.rows;
+      }
+
       const result = await cbQuery(
         `SELECT
-           d.device_id,
-           d.tenant_id,
-           d.serial_number,
-           d.created_at,
-           t.temperature,
-           t.humidity,
-           t.recorded_at,
-           t.version
+           d.device_id, d.tenant_id, d.serial_number, d.created_at,
+           t.temperature, t.humidity, t.recorded_at, t.version
          FROM device_projections d
-         LEFT JOIN device_telemetry_latest t
-           ON d.device_id = t.device_id
-         WHERE d.tenant_id = $1
+         LEFT JOIN device_telemetry_latest t ON d.device_id = t.device_id
          ORDER BY d.created_at DESC
-         LIMIT $2`,
-        [tenantId, limit]
+         LIMIT $1`,
+        [limit]
       );
+      await cache.set(cacheKey, result.rows, 30);
       return result.rows;
+    } finally {
+      await cache.releaseLock(cacheKey);
     }
-    const result = await cbQuery(
-      `SELECT
-         d.device_id,
-         d.tenant_id,
-         d.serial_number,
-         d.created_at,
-         t.temperature,
-         t.humidity,
-         t.recorded_at,
-         t.version
-       FROM device_projections d
-       LEFT JOIN device_telemetry_latest t
-         ON d.device_id = t.device_id
-       ORDER BY d.created_at DESC
-       LIMIT $1`,
-      [limit]
-    );
-    return result.rows;
   },
 
   async getDevicesWithCursor(first: number = 20, after: string | null = null, tenantId?: string) {
@@ -264,4 +296,4 @@ export const db = {
       },
     };
   },
-};
+};

apps/bff/src/datasources/redis.ts

@@ -44,4 +44,4 @@ export const cache = {
   async releaseLock(key: string): Promise<void> {
     await client.del(`lock:${key}`);
   }
-};
+};

apps/bff/src/lib/circuitBreaker.ts

@@ -142,3 +142,4 @@ export const postgresCircuitBreaker = new CircuitBreaker({
   successThreshold: 2,
   timeout:          30_000,
 });
+

apps/bff/src/pubsub.ts

@@ -4,3 +4,4 @@ export const pubsub = new PubSub();
 
 export const TELEMETRY_UPDATED = "TELEMETRY_UPDATED";
 export const TENANT_TELEMETRY_UPDATED = "TENANT_TELEMETRY_UPDATED";
+

apps/bff/src/resolvers.ts

@@ -226,3 +226,4 @@ export const resolvers = {
     },
   },
 };
+

apps/bff/src/schema.ts

@@ -90,3 +90,4 @@
     tenantTelemetryUpdated(tenantId: String!): Telemetry!
   }
 `;
+

apps/bff/src/server.ts

@@ -180,3 +180,4 @@ startServer().catch((err) => {
   console.error("Failed to start BFF:", err);
   process.exit(1);
 });
+

apps/bff/src/telemetryWatcher.ts

@@ -81,4 +81,4 @@ export async function startTelemetryWatcher() {
   });
 
   console.log("Telemetry watcher started — LISTEN telemetry_updated");
-}
+}

apps/cassandra-writer/main.go

@@ -176,3 +176,4 @@ func init() {
 	// Ensure uuid package is used
 	_ = uuid.New()
 }
+

apps/cdc-transformer/cmd/main.go

@@ -217,3 +217,4 @@ func isSkippable(err error) bool {
 	}
 	return errors.Is(err, context.Canceled)
 }
+

apps/cdc-transformer/internal/consumer/cdc_consumer.go

@@ -460,3 +460,4 @@ func isSkippable(err error) bool {
 	return errors.Is(err, context.Canceled) ||
 		strings.HasPrefix(s, "skip:")
 }
+

apps/cdc-transformer/internal/idempotency/dedupe.go

@@ -36,4 +36,4 @@ func (d *Deduper) Confirm(ctx context.Context, key string) error {
 // Cancel removes reservation when publish fails so it can retry.
 func (d *Deduper) Cancel(ctx context.Context, key string) error {
 	return d.rdb.Del(ctx, key).Err()
-}
+}

apps/cdc-transformer/internal/transform/telemetry_transform.go

@@ -155,4 +155,4 @@ func normalizeTime(t string) string {
 	}
 
 	return time.Now().UTC().Format(time.RFC3339Nano)
-}
+}

apps/cdc-transformer/internal/transform/telemetry_transform_test.go

@@ -169,3 +169,4 @@ func TestMarshalTelemetry_ProducesValidJSON(t *testing.T) {
 		t.Errorf("eventType: want telemetry.recorded got %v", result["eventType"])
 	}
 }
+

apps/dashboard/.storybook/main.ts

@@ -14,4 +14,4 @@ const config: StorybookConfig = {
   ],
   "framework": "@storybook/react-vite"
 };
-export default config;
+export default config;

apps/dashboard/.storybook/preview.ts

@@ -18,4 +18,4 @@ const preview: Preview = {
   },
 };
 
-export default preview;
+export default preview;

apps/dashboard/.storybook/vitest.setup.ts

@@ -4,4 +4,4 @@ import * as projectAnnotations from './preview';
 
 // This is an important step to apply the right configuration when testing your stories.
 // More info at: https://storybook.js.org/docs/api/portable-stories/portable-stories-vitest#setprojectannotations
-setProjectAnnotations([a11yAddonAnnotations, projectAnnotations]);
+setProjectAnnotations([a11yAddonAnnotations, projectAnnotations]);

apps/dashboard/eslint.config.js

@@ -21,3 +21,4 @@ export default defineConfig([globalIgnores(['dist']), {
     globals: globals.browser,
   },
 }, ...storybook.configs["flat/recommended"]])
+

apps/dashboard/src/features/devices/hooks/useDevices.ts

@@ -51,3 +51,4 @@ export function useSearchDevices(query: string, limit = 20) {
         error,
     };
 }
+

apps/dashboard/src/features/devices/queries/devices.graphql.ts

@@ -51,3 +51,4 @@ export const SEARCH_DEVICES = gql`
     }
   }
 `;
+

apps/dashboard/src/features/devices/types.ts

@@ -12,4 +12,4 @@ export interface DeviceTelemetryHistory {
   temperature: number | null;
   humidity: number | null;
   recordedAt: string | null;
-}
+}

apps/dashboard/src/hooks/useAuth.ts

@@ -29,3 +29,4 @@ export function useAuth() {
     getToken,
   };
 }
+

apps/dashboard/src/hooks/useDarkMode.ts

@@ -20,4 +20,4 @@ export function useDarkMode() {
   }, [isDark]);
 
   return { isDark, toggle: () => setIsDark((d) => !d) };
-}
+}

apps/dashboard/src/hooks/useDebounce.ts

@@ -14,4 +14,4 @@ export function useDebounce<T>(value: T, delay: number): T {
   }, [value, delay]);
 
   return debouncedValue;
-}
+}

apps/dashboard/src/hooks/useDeviceSubscription.ts

@@ -52,3 +52,4 @@ export function useTenantTelemetrySubscription(tenantId: string) {
     error,
   };
 }
+

apps/dashboard/src/lib/apollo.ts

@@ -105,4 +105,4 @@ const client = new ApolloClient({
   cache: new InMemoryCache(),
 });
 
-export default client;
+export default client;

apps/dashboard/src/lib/auth0.ts

@@ -8,4 +8,7 @@ export async function getAccessTokenSilently(options?: any): Promise<string> {
   if (!_getToken) throw new Error("Auth0 not initialized");
   return _getToken(options);
 }
-(window as any).__getToken = getAccessTokenSilently;
+
+if (process.env.NODE_ENV === "development") {
+  (window as any).__getToken = getAccessTokenSilently;
+}

apps/dashboard/src/stories/Button.stories.ts

@@ -52,3 +52,4 @@ export const Small: Story = {
     label: 'Button',
   },
 };
+

apps/dashboard/src/stories/Header.stories.ts

@@ -32,3 +32,4 @@ export const LoggedIn: Story = {
 };
 
 export const LoggedOut: Story = {};
+