feat: implement event sourcing architecture

backend/src/controllers/adminController.js

@@ -2,6 +2,7 @@ const { User, Trade, Position, Market } = require('../models');
 const logger = require('../config/logger');
 const stellarService = require('../services/stellarService');
 const sorobanService = require('../services/sorobanService');
+const eventSourcingService = require('../services/eventSourcingService');
 const { NotFoundError, ForbiddenError, ValidationError } = require('../middleware/errorHandler');
 
 class AdminController {
@@ -429,6 +430,33 @@ class AdminController {
       throw error;
     }
   }
+
+  // Recover all markets from event store (Event Sourcing)
+  static async recoverAllMarkets(req, res) {
+    try {
+      // Require highest admin level (double check to be safe)
+      if (req.user.userData?.level !== 'admin') {
+        throw new ForbiddenError('Only admins can initiate full market recovery');
+      }
+
+      logger.warn(`Admin ${req.user.walletAddress} initiated full market recovery`);
+
+      const result = await eventSourcingService.recoverAllMarkets();
+
+      res.json({
+        success: true,
+        message: 'Market recovery completed',
+        data: result
+      });
+    } catch (error) {
+      logger.error('Full market recovery failed:', error);
+      res.status(500).json({
+        success: false,
+        message: 'Market recovery failed',
+        error: error.message
+      });
+    }
+  }
 }
 
 module.exports = AdminController;

backend/src/controllers/marketController.js

@@ -5,6 +5,7 @@ const contractConfig = require('../config/contracts');
 const logger = require('../config/logger');
 const { NotFoundError, ValidationError, ForbiddenError, BadRequestError } = require('../middleware/errorHandler');
 const cacheService = require('../services/cacheService');
+const eventSourcingService = require('../services/eventSourcingService');
 
 class MarketController {
   // Get all markets with filtering and pagination
@@ -573,8 +574,8 @@ class MarketController {
       // Create market assets on Stellar
       const { yesAsset, noAsset, issuerKeypair } = await stellarService.createMarketAssets(marketId);
 
-      // Create market in database
-      const market = new Market({
+      // Create market via event sourcing
+      const marketPayload = {
         marketId,
         question,
         category,
@@ -586,9 +587,9 @@ class MarketController {
         noTokenAssetCode: noAsset.code,
         yesTokenIssuer: yesAsset.issuer,
         noTokenIssuer: noAsset.issuer
-      });
+      };
 
-      await market.save();
+      await eventSourcingService.appendEvent(marketId, 'MARKET_CREATED', marketPayload, req.user.walletAddress);
 
       // Update user stats
       const user = await User.findOne({ walletAddress: req.user.walletAddress });
@@ -609,20 +610,13 @@ class MarketController {
           noTokenAddress: noAsset.issuer
         });
 
-        market.metadata.contractAddress = contractResult.contractAddress;
-        await market.save();
+        await eventSourcingService.appendEvent(marketId, 'MARKET_UPDATED', { 'metadata.contractAddress': contractResult.contractAddress }, req.user.walletAddress);
       } catch (error) {
         logger.error('Failed to create Soroban contract for market:', error);
         // Continue without contract - market can still function via traditional DEX
       }
 
-      logger.market('Market created', {
-        marketId,
-        creator: req.user.walletAddress,
-        category,
-        initialLiquidity,
-        expiresAt
-      });
+      const market = await Market.findOne({ marketId });
 
       res.status(201).json({
         success: true,
@@ -664,18 +658,14 @@ class MarketController {
         return obj;
       }, {});
 
-    Object.assign(market, filteredUpdates);
-    await market.save();
-
-    logger.market('Market updated', {
-      marketId: id,
-      updater: req.user.walletAddress,
-      updates: Object.keys(filteredUpdates)
-    });
+    await eventSourcingService.appendEvent(id, 'MARKET_UPDATED', filteredUpdates, req.user.walletAddress);
+
+    // Fetch updated market to return
+    const updatedMarket = await Market.findOne({ marketId: id });
 
     res.json({
       success: true,
-      data: market,
+      data: updatedMarket,
       message: 'Market updated successfully'
     });
   }
@@ -719,12 +709,25 @@ class MarketController {
         transactionHash = resolveResult.transactionHash;
       }
 
-      // Update market in database
-      market.resolve(outcome, req.user.walletAddress, transactionHash);
+      // Update market via event sourcing
+      const resolvePayload = {
+        outcome,
+        resolvedBy: req.user.walletAddress,
+        resolutionTransactionHash: transactionHash,
+        resolvedAt: new Date()
+      };
+
       if (resolutionSource) {
-        market.metadata.resolutionSource = resolutionSource;
+        // First append resolution event
+        await eventSourcingService.appendEvent(id, 'MARKET_RESOLVED', resolvePayload, req.user.walletAddress);
+        // Then append update event for metadata
+        await eventSourcingService.appendEvent(id, 'MARKET_UPDATED', { 'metadata.resolutionSource': resolutionSource }, req.user.walletAddress);
+      } else {
+        await eventSourcingService.appendEvent(id, 'MARKET_RESOLVED', resolvePayload, req.user.walletAddress);
       }
-      await market.save();
+
+      // Fetch the updated market
+      const resolvedMarket = await Market.findOne({ marketId: id });
 
       // Update all positions for this market
       const positions = await Position.find({
@@ -750,16 +753,9 @@ class MarketController {
         }
       }
 
-      logger.market('Market resolved', {
-        marketId: id,
-        resolver: req.user.walletAddress,
-        outcome,
-        positionsUpdated: positions.length
-      });
-
       res.json({
         success: true,
-        data: market,
+        data: resolvedMarket,
         message: `Market resolved with outcome: ${outcome}`
       });
     } catch (error) {

backend/src/models/MarketEvent.js

@@ -0,0 +1,56 @@
+const mongoose = require('mongoose');
+
+const marketEventSchema = new mongoose.Schema({
+  marketId: {
+    type: String,
+    required: true,
+    index: true
+  },
+  eventType: {
+    type: String,
+    required: true,
+    enum: [
+      'MARKET_CREATED',
+      'MARKET_UPDATED',
+      'MARKET_RESOLVED',
+      'MARKET_CANCELLED',
+      'MARKET_ARCHIVED',
+      'TRADE_EXECUTED',
+      'PRICES_UPDATED',
+      'LIQUIDITY_ADDED',
+      'LIQUIDITY_REMOVED'
+    ],
+    index: true
+  },
+  schemaVersion: {
+    type: Number,
+    required: true,
+    default: 1
+  },
+  payload: {
+    type: mongoose.Schema.Types.Mixed,
+    required: true
+  },
+  actorAddress: {
+    type: String,
+    index: true
+  },
+  timestamp: {
+    type: Date,
+    default: Date.now,
+    index: true
+  },
+  sequenceNumber: {
+    type: Number,
+    required: true,
+    index: true
+  }
+}, {
+  timestamps: true,
+  collection: 'market_events'
+});
+
+// Ensure sequence numbers are unique per market
+marketEventSchema.index({ marketId: 1, sequenceNumber: 1 }, { unique: true });
+
+module.exports = mongoose.model('MarketEvent', marketEventSchema);

backend/src/models/MarketSnapshot.js

@@ -0,0 +1,28 @@
+const mongoose = require('mongoose');
+
+const marketSnapshotSchema = new mongoose.Schema({
+  marketId: {
+    type: String,
+    required: true,
+    index: true,
+    unique: true
+  },
+  lastEventSequenceNumber: {
+    type: Number,
+    required: true,
+    default: 0
+  },
+  stateData: {
+    type: mongoose.Schema.Types.Mixed,
+    required: true
+  },
+  timestamp: {
+    type: Date,
+    default: Date.now
+  }
+}, {
+  timestamps: true,
+  collection: 'market_snapshots'
+});
+
+module.exports = mongoose.model('MarketSnapshot', marketSnapshotSchema);

backend/src/models/index.js

@@ -4,6 +4,8 @@ const Trade = require('./Trade');
 const Position = require('./Position');
 const IndexedEvent = require('./IndexedEvent');
 const ResolutionEvent = require('./ResolutionEvent');
+const MarketEvent = require('./MarketEvent');
+const MarketSnapshot = require('./MarketSnapshot');
 const Alert = require('./Alert');
 const EventSchedule = require('./EventSchedule');
 const WhaleTransaction = require('./WhaleTransaction');
@@ -22,6 +24,8 @@ module.exports = {
   Position,
   IndexedEvent,
   ResolutionEvent,
+  MarketEvent,
+  MarketSnapshot,
   Alert,
   EventSchedule,
   WhaleTransaction,

backend/src/routes/admin.js

@@ -30,6 +30,12 @@ router.put('/markets/:marketId/resolve',
   asyncHandler(adminController.forceResolveMarket)
 );
 
+// Recover all markets from event store (Event Sourcing)
+router.post('/markets/recover',
+  requireAdmin,
+  asyncHandler(adminController.recoverAllMarkets)
+);
+
 // Get system logs
 router.get('/logs',
   requireAdmin,

backend/src/services/contractEventIndexer.js

@@ -2,6 +2,7 @@
 const sorobanService = require('./sorobanService');
 const contractConfig = require('../config/contracts');
 const { Market, Trade, Position, User, IndexedEvent, ResolutionEvent } = require('../models');
+const eventSourcingService = require('./eventSourcingService');
 
 class ContractEventIndexer {
   constructor() {
@@ -323,8 +324,8 @@
         return;
       }
 
-      // Create new market record
-      const market = new Market({
+      // Create new market record via event sourcing
+      const marketPayload = {
         marketId,
         question,
         category,
@@ -343,9 +344,9 @@
         currentNoPrice: 0.5,
         blockchainTxHash: metadata.txHash,
         createdAt: new Date()
-      });
+      };
 
-      await market.save();
+      await eventSourcingService.appendEvent(marketId, 'MARKET_CREATED', marketPayload, creator);
       logger.info(`Indexed new market: ${marketId}`, { question });
     } catch (error) {
       logger.error('Failed to handle market created event:', error);
@@ -427,15 +428,11 @@
     try {
       const { marketId, outcome, resolvedAt } = eventValue;
 
-      await Market.findOneAndUpdate(
-        { marketId },
-        {
-          status: 'resolved',
-          resolvedOutcome: outcome,
-          resolvedAt: new Date(resolvedAt * 1000),
-          resolutionTxHash: metadata.txHash
-        }
-      );
+      await eventSourcingService.appendEvent(marketId, 'MARKET_RESOLVED', {
+        outcome,
+        resolvedAt: new Date(resolvedAt * 1000),
+        resolutionTransactionHash: metadata.txHash
+      }, 'SYSTEM_INDEXER');
 
       await this.updateReputationFromResolvedMarket(marketId, outcome);
 
@@ -574,21 +571,24 @@
    */
   async updateMarketStats(marketId, trade) {
     try {
-      const update = {
-        $inc: {
-          totalVolume: trade.totalCost,
-          totalTrades: 1
-        }
-      };
+      const amount = trade.totalCost; // or trade.amount based on logic
+      await eventSourcingService.appendEvent(marketId, 'TRADE_EXECUTED', { amount }, trade.userWalletAddress);
 
       // Update current prices based on latest trades
+      const pricePayload = {};
+      const currentMarket = await Market.findOne({ marketId });
+      let yesPrice = currentMarket ? currentMarket.currentYesPrice : 0.5;
+      let noPrice = currentMarket ? currentMarket.currentNoPrice : 0.5;
+
       if (trade.tokenType === 'yes') {
-        update.currentYesPrice = trade.price;
+        yesPrice = trade.price;
+        noPrice = 1.0 - trade.price;
       } else if (trade.tokenType === 'no') {
-        update.currentNoPrice = trade.price;
+        noPrice = trade.price;
+        yesPrice = 1.0 - trade.price;
       }
 
-      await Market.findOneAndUpdate({ marketId }, update);
+      await eventSourcingService.appendEvent(marketId, 'PRICES_UPDATED', { yesPrice, noPrice }, trade.userWalletAddress);
     } catch (error) {
       logger.error('Failed to update market stats:', error);
     }