Anjing Webshells

Website : rimsha.abasa.com
: / var / www / mudeerapi.abasa.com / nodetest-backup30April26 / src /
Filename : index.js
back
import dotenv from "dotenv";
import http from 'http';
import { app } from './app.js';
import connectDB from "./db/index.js";
import { MongoClient } from "mongodb";

dotenv.config({
    path: `.env`
});
import connectRedis from './db/redis.js';
import mongoose from "mongoose";
import { setupUserSockets } from './sockets/userSocket.js';
import { migrateToTenureSystem } from './controllers/leave.controller.js';
import { migrateAttendanceActivity } from './scripts/migrateAttendanceActivity.js';


connectDB()
    .then(() => {
        const httpServer = http.createServer(app);
        
        httpServer.listen(process.env.PORT || 8000, '0.0.0.0', () => {
            console.log(`⚙️ Server is running at port : ${process.env.PORT}`);
        });
        
        // Setup WebSocket server
        setupUserSockets(httpServer);


        backfillJoinedOn().catch((err) => {
            console.error('backfillJoinedOn failed:', err.message);
        });

        migrateToTenureSystem()
            .then(({ totalUsers, results }) => {
                const errors = results.filter((r) => !r.success);
                if (errors.length > 0) {
                    console.warn(
                        `migrateToTenureSystem: ${totalUsers} user(s), ${errors.length} error(s). First: ${errors[0]?.error ?? 'unknown'}`
                    );
                } else {
                    console.log(`migrateToTenureSystem: synced leave counters to UserTenure for ${totalUsers} user(s) (all tenure windows from join date) and dropped legacy user fields.`);
                }
            })
            .catch((err) => {
                console.error('migrateToTenureSystem failed:', err?.message ?? err);
            });


            migrateAttendanceActivity({ closeConnection: false })
            .then(({ processed, updated }) => {
                console.log(`migrateAttendanceActivity: scanned ${processed}, updated ${updated}.`);
            })
            .catch((err) => {
                console.error('migrateAttendanceActivity failed:', err?.message ?? err);
            });
        

        // if(process.env.NODE_ENV === 'STAGING') {
        //     replicateChanges();
        // }
    })
    .catch((err) => {
        console.log("MONGO db connection failed !!! ", err);
    });


await connectRedis();



const DEFAULT_JOINED_ON = new Date('2026-01-01T00:00:00.000Z');

async function backfillJoinedOn() {
    try {
        const collection = mongoose.connection.collection('users');
        const result = await collection.updateMany(
            { $or: [{ joined_on: { $exists: false } }, { joined_on: null }] },
            { $set: { joined_on: DEFAULT_JOINED_ON } }
        );
        if (result.modifiedCount > 0) {
            console.log(`backfillJoinedOn: set joined_on on ${result.modifiedCount} user(s).`);
        }
    } catch (error) {
        console.error('Error backfilling joined_on:', error.message);
        throw error;
    }
}

async function replicateChanges() {
    const sourceUri = process.env.SOURCE_URI;
    const targetUri = process.env.DATABASE_URL;

    // Removed deprecated useUnifiedTopology option
    const sourceClient = new MongoClient(sourceUri);
    const targetClient = new MongoClient(targetUri);

    // merna nadia 
    try {
        // Connect to both clusters
        await sourceClient.connect();
        await targetClient.connect();

        const sourceDb = sourceClient.db();
        const targetDb = targetClient.db();

        // Verify source database supports change streams (must be replica set)
        try {
            const adminDb = sourceClient.db('admin');
            const serverStatus = await adminDb.command({ replSetGetStatus: 1 });
            console.log(`Source database is a replica set: ${serverStatus.set}`);
        } catch (rsError) {
            console.error('Change streams require a replica set. Source database may not be configured as a replica set.');
            console.error('Skipping replication setup. Error:', rsError.message);
            await sourceClient.close();
            await targetClient.close();
            return;
        }

        // Get user IDs from target database
        const targetUsers = await targetDb.collection('users').find({}).project({ _id: 1 }).toArray();
        const targetUserIds = targetUsers.map(user => user._id);
        const targetUserIdsAsStrings = targetUserIds.map(id => id.toString());
        
        if (targetUserIds.length === 0) {
            console.log('No users found in target database. Skipping replication setup.');
            return;
        }
        
        console.log(`Watching changes for ${targetUserIds.length} users from target database`);

        // Collections to watch (user-specific collections)
        // eventactivities uses ObjectId, activewindows and attendances use String
        const collectionsWithObjectId = ['eventactivities'];
        const collectionsWithString = ['activewindows', 'attendances'];

        // Helper function to setup change stream with error handling
        const setupChangeStream = (sourceCollection, targetCollection, collectionName, userIdFilter, isObjectId) => {
            const changeStream = sourceCollection.watch(
                [
                    {
                        $match: {
                            $or: [
                                { 'fullDocument.user_id': { $in: userIdFilter } },
                                { operationType: 'delete' }
                            ]
                        }
                    }
                ],
                { fullDocument: 'updateLookup' }
            );

            // Handle change stream errors gracefully
            changeStream.on('error', (error) => {
                console.error(`Change stream error on ${collectionName}:`, error.message);
                // Don't crash the server, just log the error
            });

            changeStream.on('change', async (change) => {
                try {
                    // For delete operations, check if document exists in target before deleting
                    if (change.operationType === 'delete') {
                        const deletedDoc = await targetCollection.findOne({ _id: change.documentKey._id });
                        if (deletedDoc) {
                            await targetCollection.deleteOne({ _id: change.documentKey._id });
                            console.log(`Replicated delete operation on ${collectionName} collection`);
                        }
                        return;
                    }

                    const document = change.fullDocument;
                    if (!document) return;

                    // Verify user_id matches
                    const userId = document.user_id;
                    let isTargetUser;
                    
                    if (isObjectId) {
                        isTargetUser = userIdFilter.some(id => id.equals(userId));
                    } else {
                        isTargetUser = userIdFilter.includes(userId);
                    }
                    
                    if (!isTargetUser) {
                        return; // Skip if not for our target users
                    }

                    await targetCollection.updateOne(
                        { _id: document._id }, 
                        { $set: document }, 
                        { upsert: true }
                    );
                    console.log(`Replicated ${change.operationType} operation on ${collectionName} collection`);
                } catch (error) {
                    console.error(`Error during replication on ${collectionName}:`, error.message);
                }
            });

            console.log(`Watching changes on ${collectionName} for target users...`);
        };

        // Watch change streams for collections with ObjectId user_id
        collectionsWithObjectId.forEach((collectionName) => {
            const sourceCollection = sourceDb.collection(collectionName);
            const targetCollection = targetDb.collection(collectionName);
            setupChangeStream(sourceCollection, targetCollection, collectionName, targetUserIds, true);
        });

        // Watch change streams for collections with String user_id
        collectionsWithString.forEach((collectionName) => {
            const sourceCollection = sourceDb.collection(collectionName);
            const targetCollection = targetDb.collection(collectionName);
            setupChangeStream(sourceCollection, targetCollection, collectionName, targetUserIdsAsStrings, false);
        });

    } catch (err) {
        console.error('Error in replicateChanges:', err.message);
    }
}