Feat/graphql ws collab 1123

backend/package.json

@@ -16,6 +16,8 @@
     "validate:curriculum": "tsx scripts/validate-curriculum.ts",
     "bench": "tsx benchmarks/runBenchmarks.ts",
     "collaboration": "tsx src/collaborationServer.ts",
+    "collab:graphql": "tsx src/graphqlWsServer.ts",
+    "load:ws": "node scripts/ws-load-test.js",
     "db:seed": "tsx prisma/seed.ts"
   },
   "prisma": {
@@ -60,6 +62,7 @@
     "yaml": "^2.0.0",
     "y-websocket": "^3.0.0",
     "yjs": "^13.6.30",
+    "graphql-ws": "^6.2.2",
     "zod": "^4.3.6"
   },
   "devDependencies": {

backend/scripts/ws-load-test.js

@@ -0,0 +1,40 @@
+const autocannon = require('autocannon');
+
+const url = process.env.TEST_URL || 'http://localhost:4001/graphql';
+const connections = Number(process.env.CONNS || 1000);
+const duration = Number(process.env.DURATION || 10);
+
+console.log(`Starting WS load test against ${url} with ${connections} connections for ${duration}s`);
+
+autocannon({
+  url,
+  connections,
+  duration,
+  pipelining: 1,
+  timeout: 20,
+  headers: { Connection: 'Upgrade', Upgrade: 'websocket' },
+  setupClient: (client) => {
+    client.on('connect', () => {
+      // send a simple graphql-ws connection_init
+      client.send(JSON.stringify({ type: 'connection_init', payload: {} }));
+      // send a small subscribe payload for courseUpdated with random id
+      const id = Math.random().toString(36).slice(2, 8);
+      const payload = {
+        id,
+        type: 'start',
+        payload: {
+          query: 'subscription($courseId: ID!){ courseUpdated(courseId: $courseId){ id title } }',
+          variables: { courseId: 'test-room' },
+        },
+      };
+      client.send(JSON.stringify(payload));
+    });
+    client.on('data', () => {});
+  },
+}, (err, res) => {
+  if (err) {
+    console.error('autocannon error', err);
+    process.exit(1);
+  }
+  console.log('Finished', res);
+});

backend/src/config/env.config.ts

@@ -34,6 +34,12 @@ export const config = {
   db: {
     url: getEnvVar('DATABASE_URL'),
     readReplicaUrl: getEnvVar('DATABASE_READ_REPLICA_URL', ''),
+    replica: {
+      checkIntervalMs: parseInt(getEnvVar('DB_REPLICA_CHECK_INTERVAL_MS', '10000'), 10),
+      failureThreshold: parseInt(getEnvVar('DB_REPLICA_FAILURE_THRESHOLD', '3'), 10),
+      cooldownMs: parseInt(getEnvVar('DB_REPLICA_COOLDOWN_MS', '30000'), 10),
+      replicationLagWindowMs: parseInt(getEnvVar('DB_REPLICATION_LAG_WINDOW_MS', '1000'), 10),
+    },
   },
   redis: {
     url: getEnvVar('REDIS_URL'), // Required

backend/src/db/index.ts

@@ -1,7 +1,7 @@
 import { PrismaClient } from '@prisma/client';
 import config from '../config/env.config.js';
+import { encryptionMiddleware } from '../middleware/prismaEncryption.js';
 import { getWorkspaceId } from '../middleware/WorkspaceContext.js';
-import { getDatabaseRoleForOperation } from './requestContext.js';
 import logger from '../utils/logger.js';
 import { encryptionMiddleware } from '../middleware/prismaEncryption.js';
 import { workspaceModels } from './workspaceModels.js';
@@ -117,14 +117,85 @@ const workspaceExtension = {
 const prisma = basePrisma.$extends(workspaceExtension);
 const readPrisma = baseReadPrisma.$extends(workspaceExtension);
 
+// Read-replica health & circuit breaker state
+let readReplicaHealthy = true;
+let replicaFailureCount = 0;
+let replicaCircuitOpenUntil = 0;
+
+const getReplicaFailureThreshold = () => config.db?.replica?.failureThreshold ?? 3;
+const getReplicaCooldownMs = () => config.db?.replica?.cooldownMs ?? 30000;
+const getReplicaCheckIntervalMs = () => config.db?.replica?.checkIntervalMs ?? 10000;
+const getReplicaLagWindowMs = () => config.db?.replica?.replicationLagWindowMs ?? 1000;
+
+const markReadReplicaFailure = () => {
+  replicaFailureCount += 1;
+  if (replicaFailureCount >= getReplicaFailureThreshold()) {
+    readReplicaHealthy = false;
+    replicaCircuitOpenUntil = Date.now() + getReplicaCooldownMs();
+    logger.warn('Read replica circuit opened', { until: replicaCircuitOpenUntil });
+  }
+};
+
+const probeReadReplica = async () => {
+  try {
+    await readPool.query('SELECT 1');
+    // success
+    replicaFailureCount = 0;
+    readReplicaHealthy = true;
+    replicaCircuitOpenUntil = 0;
+    logger.info('Read replica probe succeeded, circuit closed');
+  } catch (err) {
+    replicaCircuitOpenUntil = Date.now() + getReplicaCooldownMs();
+    logger.warn('Read replica probe failed', { err });
+  }
+};
+
+const checkReadReplicaHealth = async () => {
+  // If circuit is open, avoid frequent checks until cooldown expires
+  if (!readReplicaHealthy && Date.now() < replicaCircuitOpenUntil) return;
+
+  try {
+    const res = await readPool.query(
+      "SELECT EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp())) * 1000 AS lag_ms"
+    );
+    const lagMs = res?.rows?.[0]?.lag_ms;
+    const lagWindow = getReplicaLagWindowMs();
+
+    if (lagMs === null || lagMs === undefined) {
+      throw new Error('pg_last_xact_replay_timestamp returned null (replica may not be streaming)');
+    }
+
+    if (Number(lagMs) > lagWindow) {
+      throw new Error(`replica lag ${lagMs}ms exceeds window ${lagWindow}ms`);
+    }
+
+    // healthy
+    replicaFailureCount = 0;
+    readReplicaHealthy = true;
+  } catch (err) {
+    logger.warn('Read replica health check failed', { err: String(err), failureCount: replicaFailureCount });
+    markReadReplicaFailure();
+  }
+
+  // If circuit was open and cooldown expired, probe once
+  if (!readReplicaHealthy && Date.now() >= replicaCircuitOpenUntil) {
+    await probeReadReplica();
+  }
+};
+
+// Start periodic health checks
+setInterval(() => {
+  void checkReadReplicaHealth();
+}, getReplicaCheckIntervalMs());
+
 const routingExtension = {
   name: 'read-replica-routing',
   query: {
     $allModels: {
       async $allOperations({ model, operation, args, query }: { model?: string; operation?: string; args?: any; query: (args: any) => Promise<any> }) {
           const dbRole = getDatabaseRoleForOperation(operation!);
 
-        if (dbRole === 'read') {
+        if (dbRole === 'read' && readReplicaHealthy) {
           const modelClient = readPrisma[model as keyof typeof readPrisma];
           if (modelClient && typeof modelClient[operation as keyof typeof modelClient] === 'function') {
             try {
@@ -134,8 +205,16 @@ const routingExtension = {
                 `Read replica query failed for ${model}.${operation}, falling back to primary:`,
                 error
               );
+              // mark failure for circuit breaker
+              try {
+                markReadReplicaFailure();
+              } catch (e) {
+                logger.warn('Failed to mark read replica failure', { err: String(e) });
+              }
             }
           }
+        } else if (dbRole === 'read' && !readReplicaHealthy) {
+          logger.debug('Skipping read replica route because replica circuit is open; using primary');
         }
 
         const modelClient = prisma[model as keyof typeof prisma];
@@ -169,4 +248,4 @@ if (process.env.NODE_ENV !== 'production') {
 }
 
 export { prisma, readPrisma };
-export default routedPrisma as PrismaClient;
+export default routedPrisma as PrismaClient;

backend/src/graphqlWsServer.ts

@@ -0,0 +1,151 @@
+import express from 'express';
+import { buildSchema, execute, subscribe } from 'graphql';
+import { useServer } from 'graphql-ws/lib/use/ws';
+import http from 'http';
+import jwt from 'jsonwebtoken';
+import { WebSocketServer } from 'ws';
+import { typeDefs } from './graphql/schema.js';
+import { graphQLMiddleware } from './graphql/server.js';
+
+const PORT = Number(process.env.PORT || process.env.WS_PORT || 4001);
+const JWT_SECRET = process.env.JWT_SECRET || 'dev-secret';
+
+class PubSub {
+  private events = new Map<string, Set<(payload: any) => void>>();
+
+  publish(topic: string, payload: any) {
+    const subs = this.events.get(topic);
+    if (!subs) return;
+    for (const cb of subs) cb(payload);
+  }
+
+  subscribe(topic: string) {
+    const set = this.events.get(topic) ?? new Set();
+    this.events.set(topic, set);
+    const queue: any[] = [];
+    let pullResolve: ((v: IteratorResult<any>) => void) | null = null;
+
+    const push = (value: any) => {
+      if (pullResolve) {
+        pullResolve({ value, done: false });
+        pullResolve = null;
+      } else {
+        queue.push(value);
+      }
+    };
+
+    const cb = (payload: any) => push(payload);
+    set.add(cb);
+
+    const asyncIterator = {
+      async next() {
+        if (queue.length) return { value: queue.shift(), done: false };
+        return await new Promise<IteratorResult<any>>(res => (pullResolve = res));
+      },
+      return() {
+        set.delete(cb);
+        return Promise.resolve({ value: undefined, done: true });
+      },
+      throw(error: any) {
+        return Promise.reject(error);
+      },
+      [Symbol.asyncIterator]() {
+        return this;
+      },
+    };
+
+    return asyncIterator;
+  }
+}
+
+const pubsub = new PubSub();
+
+const schema = buildSchema(typeDefs);
+
+const rootValue: Record<string, any> = {
+  courseUpdated: ({ courseId }: { courseId: string }) => pubsub.subscribe(`course:${courseId}`),
+};
+
+const app = express();
+
+app.get('/health', (_req, res) => res.status(200).send('ok'));
+
+(async () => {
+  const graphqlHandlers = await graphQLMiddleware();
+  // mount middleware array returned by graphQLMiddleware at /graphql
+  app.use('/graphql', ...graphqlHandlers);
+
+  const server = http.createServer(app);
+
+  const wsServer = new WebSocketServer({ server, path: '/graphql' });
+
+  // per-client backpressure queue limit
+  const QUEUE_LIMIT = 32;
+
+  useServer(
+    {
+      schema,
+      execute,
+      subscribe,
+      rootValue,
+      onConnect: async (ctx) => {
+        const connectionParams = (ctx.connectionParams || {}) as Record<string, any>;
+        const token = (connectionParams.authorization || connectionParams.token || '').replace(/^Bearer\s+/i, '');
+        if (!token) throw new Error('Missing auth token');
+        try {
+          const user = jwt.verify(token, JWT_SECRET);
+          return { user };
+        } catch (err) {
+          throw new Error('Unauthorized');
+        }
+      },
+      onSubscribe: async (ctx, msg) => {
+        // multiplexing: allow clients to subscribe to different rooms via variables
+        return msg.payload;
+      },
+      onNext: (ctx, msg, args) => {
+        // no-op: handled by graphql execution
+      },
+      onUnhandledError: (ctx, err) => {
+        console.error('WS error', err);
+      },
+      onClose: (ctx, code, reason) => {
+        // cleanup handled by graphql-ws
+      },
+      // customize sending to add backpressure dropping stale frames
+      sendMessage: (socket, message) => {
+        // attach a small queue on the socket
+        const qSymbol = Symbol.for('gqlws_queue');
+        // @ts-ignore
+        if (!socket[qSymbol]) socket[qSymbol] = [];
+        // @ts-ignore
+        const q = socket[qSymbol] as any[];
+        q.push(message);
+        if (q.length > QUEUE_LIMIT) {
+          // drop oldest
+          q.shift();
+        }
+        // if socket is ready, flush
+        // @ts-ignore
+        if (socket.readyState === socket.OPEN) {
+          // flush all
+          // @ts-ignore
+          while (q.length) socket.send(JSON.stringify(q.shift()));
+        }
+        return Promise.resolve();
+      },
+    },
+    wsServer
+  );
+
+  server.listen(PORT, () => console.log(`GraphQL WS server listening on ${PORT}`));
+
+  // expose a tiny API to publish to rooms (used by other parts of the app)
+  app.post('/publish/:room', express.json(), (req, res) => {
+    const room = req.params.room;
+    pubsub.publish(room, req.body.payload ?? req.body);
+    res.status(204).end();
+  });
+})();
+
+export { };

backend/src/index.ts

@@ -1,12 +1,10 @@
 
-import express, { Request, Response } from 'express';
 import dotenv from 'dotenv';
-import routes from './routes/index.js';
-import { initializeSentry, getSentryRequestHandler, getSentryErrorHandler } from './utils/sentry.js';
-import { jsonBodySizeLimit } from './middleware/bodySizeLimit.js';
+import express, { Request, Response } from 'express';
 import { createCorsMiddleware } from './config/cors.config.js';
 import swaggerDocsRouter from './config/swagger.serve.js';
 import logger from './utils/logger.js';
+import { getSentryErrorHandler, getSentryRequestHandler, initializeSentry } from './utils/sentry.js';
 
 
 dotenv.config();
@@ -25,6 +23,8 @@ app.use(createCorsMiddleware());
 app.use(express.json());
 app.use(express.urlencoded({ extended: true }));
 app.use(jsonBodySizeLimit);
+// Attach DB routing context middleware to ensure GETs are routed to replicas
+app.use(dbRoutingMiddleware);
 
 app.get('/health', (req: Request, res: Response) => {
   res.json({ status: 'ok', message: 'Web3 Student Lab Backend is running' });

frontend/data/contract-registry.json

@@ -0,0 +1,15 @@
+{
+  "contracts": [
+    {
+      "id": "example-token",
+      "network": "testnet",
+      "address": "GCEXAMPLETOKENADDRESS",
+      "expectedHash": ""
+    }
+  ],
+  "rpcs": {
+    "testnet": ["https://soroban-testnet.example/rpc"],
+    "futurenet": ["https://soroban-futurenet.example/rpc"],
+    "mainnet": ["https://soroban-mainnet.example/rpc"]
+  }
+}

frontend/data/moderation-audit.json

@@ -0,0 +1,3 @@
+{
+  "audit": []
+}