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
+

Makefile

@@ -0,0 +1,103 @@
+.PHONY: up down restart logs build seed test lint clean help
+
+# ============================================
+# GrainGuard — Developer Makefile
+# ============================================
+
+help: ## Show this help
+	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}'
+
+# ============================================
+# Docker Compose
+# ============================================
+
+up: ## Start all services
+	docker compose -f infra/docker/docker-compose.yml up -d
+
+down: ## Stop all services
+	docker compose -f infra/docker/docker-compose.yml down
+
+restart: ## Restart all services
+	docker compose -f infra/docker/docker-compose.yml restart
+
+logs: ## Tail logs for all services
+	docker compose -f infra/docker/docker-compose.yml logs -f
+
+logs-gateway: ## Tail gateway logs
+	docker compose -f infra/docker/docker-compose.yml logs -f gateway
+
+logs-bff: ## Tail BFF logs
+	docker compose -f infra/docker/docker-compose.yml logs -f bff
+
+logs-kafka: ## Tail Kafka logs
+	docker compose -f infra/docker/docker-compose.yml logs -f kafka
+
+build: ## Build all Docker images
+	docker compose -f infra/docker/docker-compose.yml build
+
+# ============================================
+# Database
+# ============================================
+
+seed: ## Seed dev data (Postgres + Elasticsearch)
+	cd scripts/seed && bash seed-postgres.sh
+	cd scripts/seed && bash seed-elasticsearch.sh
+
+migrate: ## Run database migrations
+	go run libs/migrate/migrate.go
+
+# ============================================
+# Testing
+# ============================================
+
+test: ## Run all tests
+	$(MAKE) test-go
+	$(MAKE) test-react
+
+test-go: ## Run Go tests
+	go test ./...
+
+test-react: ## Run React/Node tests
+	cd apps/dashboard && npm test -- --run
+	cd apps/bff && npm test -- --run
+
+test-load: ## Run k6 load tests
+	k6 run scripts/load-tests/gateway-load-test.js
+	k6 run scripts/load-tests/bff-load-test.js
+
+# ============================================
+# Linting
+# ============================================
+
+lint: ## Lint all services
+	$(MAKE) lint-go
+	$(MAKE) lint-ts
+
+lint-go: ## Lint Go services
+	golangci-lint run ./...
+
+lint-ts: ## Lint TypeScript services
+	cd apps/gateway && npm run lint
+	cd apps/bff && npm run lint
+	cd apps/dashboard && npm run lint
+
+# ============================================
+# Cleanup
+# ============================================
+
+clean: ## Remove all containers and volumes
+	docker compose -f infra/docker/docker-compose.yml down -v --remove-orphans
+
+clean-cache: ## Clear Redis cache
+	docker compose -f infra/docker/docker-compose.yml exec redis redis-cli FLUSHALL
+
+# ============================================
+# Status
+# ============================================
+
+ps: ## Show running services
+	docker compose -f infra/docker/docker-compose.yml ps
+
+health: ## Check health of all services
+	curl -s http://localhost:8086/health | jq
+	curl -s http://localhost:4000/health | jq

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()
-}
+}