index.ts

import dayjs from 'dayjs';
import { MongoClient } from 'mongodb';
import { Decimal128, Double } from 'bson';
import { IAdminForthDataSourceConnector, IAdminForthSingleFilter, IAdminForthAndOrFilter, AdminForthResource, IAggregationRule, IGroupByRule, IGroupByDateTrunc, IGroupByField } from 'adminforth';
import { afLogger, checkIfFieldIsInsideResourceColumns } from 'adminforth';
import { AdminForthDataTypes, AdminForthFilterOperators, AdminForthSortDirections, AdminForthBaseConnector } from 'adminforth';

const escapeRegex = (value: any) => {
    return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); // Escapes special characters
};
function normalizeMongoValue(v: any) {
  if (v == null) {
    return v;
  }
  if (v instanceof Decimal128) {
    return v.toString();
  }
  if (v instanceof Double) {
    return v.valueOf();
  }
  if (typeof v === "object" && v.$numberDecimal) {
    return String(v.$numberDecimal);
  }
  if (typeof v === "object" && v.$numberDouble) {
    return Number(v.$numberDouble);
  }
  return v;
}

class MongoConnector extends AdminForthBaseConnector implements IAdminForthDataSourceConnector {

    async setupClient(url: any): Promise<void> {
        this.client = new MongoClient(url);
        (async () => {
            try {
                await this.client.connect();
                this.client.on('error', (err: any) => {
                    afLogger.error(`Mongo error: ${err.message}`);
                });
                afLogger.info('Connected to Mongo');
            } catch (e) {
                afLogger.error(`Failed to connect to Mongo: ${e}`);
            }
        })();
    }

    OperatorsMap = {
        [AdminForthFilterOperators.EQ]: (value: any) => value,
        [AdminForthFilterOperators.NE]: (value: any) => ({ $ne: value }),
        [AdminForthFilterOperators.GT]: (value: any) => ({ $gt: value }),
        [AdminForthFilterOperators.LT]: (value: any) => ({ $lt: value }),
        [AdminForthFilterOperators.GTE]: (value: any) => ({ $gte: value }),
        [AdminForthFilterOperators.LTE]: (value: any) => ({ $lte: value }),
        [AdminForthFilterOperators.LIKE]: (value: any) => ({ $regex: escapeRegex(value) }),
        [AdminForthFilterOperators.ILIKE]: (value: any) => ({ $regex: escapeRegex(value), $options: 'i' }),
        [AdminForthFilterOperators.IN]: (value: any) => ({ $in: value }),
        [AdminForthFilterOperators.NIN]: (value: any) => ({ $nin: value }),
        [AdminForthFilterOperators.AND]: (value: any) => ({ $and: value }),
        [AdminForthFilterOperators.OR]: (value: any) => ({ $or: value }),
        [AdminForthFilterOperators.IS_EMPTY]: () => null,
        [AdminForthFilterOperators.IS_NOT_EMPTY]: () => ({ $ne: null }),
    };

    SortDirectionsMap = {
        [AdminForthSortDirections.asc]: 1,
        [AdminForthSortDirections.desc]: -1,
    };
    async getAllTables(): Promise<Array<string>>{
        const db = this.client.db();
    
        const collections = await db.listCollections().toArray();
    
        return collections.map((col: any) => col.name);
    }

    async getAllColumnsInTable(collectionName: string): Promise<Array<{ name: string; type: string; isPrimaryKey?: boolean; sampleValue?: any; }>> {

        const sampleDocs = await this.client.db().collection(collectionName).find({}).sort({ _id: -1 }).limit(100).toArray();
      
        const fieldTypes = new Map<string, Set<string>>();
        const sampleValues = new Map<string, any>();

        function detectType(value: any): string {
          if (value === null || value === undefined) return 'string';
          if (typeof value === 'string') return 'string';
          if (typeof value === 'boolean') return 'boolean';
          if (typeof value === 'number') {
            return Number.isInteger(value) ? 'integer' : 'float';
          }
          if (value instanceof Date) return 'datetime';
          if (value && typeof value === 'object' && ('$numberDecimal' in value || value._bsontype === 'Decimal128')) return 'decimal';
          if (typeof value === 'object') return 'json';
          return 'string';
        }
      
        function addType(name: string, type: string) {
          if (!fieldTypes.has(name)) {
            fieldTypes.set(name, new Set());
          }
          fieldTypes.get(name)!.add(type);
        }
      
        function flattenObject(obj: any, prefix = '') {
          Object.entries(obj).forEach(([key, value]) => {
            const fullKey = prefix ? `${prefix}.${key}` : key;

            if (!fieldTypes.has(fullKey)) {
              fieldTypes.set(fullKey, new Set());
              sampleValues.set(fullKey, value);
            }
      
            if (value instanceof Buffer) {
              addType(fullKey, 'json');
              return;
            }
            if (
                value &&
                typeof value === 'object' &&
                ('$numberDecimal' in value || (value as any)._bsontype === 'Decimal128')
              ) {
              addType(fullKey, 'decimal');
              return;
            }
      
            if (
              value && 
              typeof value === 'object' && 
              !Array.isArray(value) && 
              !(value instanceof Date)
            ) {
              addType(fullKey, 'json');
              return
            }
        
            addType(fullKey, detectType(value));
          });
        }
      
        for (const doc of sampleDocs) {
          flattenObject(doc);
        }
      
        return Array.from(fieldTypes.entries()).map(([name, types]) => {
            const primaryKey = name === '_id';
        
            const priority = ['datetime', 'date', 'decimal', 'integer', 'float', 'boolean', 'json', 'string'];
        
            const matched = priority.find(t => types.has(t)) || 'string';
        
            const typeMap: Record<string, AdminForthDataTypes> = {
                string: AdminForthDataTypes.STRING,
                integer: AdminForthDataTypes.INTEGER,
                float: AdminForthDataTypes.FLOAT,
                boolean: AdminForthDataTypes.BOOLEAN,
                datetime: AdminForthDataTypes.DATETIME,
                date: AdminForthDataTypes.DATE,
                json: AdminForthDataTypes.JSON,
                decimal: AdminForthDataTypes.DECIMAL,
            };
            return {
                name,
                type: typeMap[matched] ?? AdminForthDataTypes.STRING,
                ...(primaryKey ? { isPrimaryKey: true } : {}),
                sampleValue: sampleValues.get(name),
            };
        });
    }
      
    
    async discoverFields(resource: any) {
        return resource.columns.filter((col: any) => !col.virtual).reduce((acc: any, col: any) => {
            if (!col.type) {
                throw new Error(`Type is not defined for column ${col.name} in resource ${resource.table}`);
            }

            acc[col.name] = {
                name: col.name,
                type: col.type,
                primaryKey: col.primaryKey,
                virtual: col.virtual,
                _underlineType: col._underlineType,
            };
            return acc;
        }, {});
    }

    getPrimaryKey(resource: any) {
        for (const col of resource.dataSourceColumns) {
            if (col.primaryKey) {
                return col.name;
            }
        }
    }

    getFieldValue(field: any, value: any) {
        if (field.type == AdminForthDataTypes.DATETIME) {
            if (!value) {
                return null;
            }
            return dayjs(Date.parse(value)).toISOString();

        } else if (field.type == AdminForthDataTypes.DATE) {
            if (!value) {
                return null;
            }
            return dayjs(Date.parse(value)).toISOString().split('T')[0];

        } else if (field.type == AdminForthDataTypes.BOOLEAN) {
          return value === null ? null : !!value;
        } else if (field.type == AdminForthDataTypes.DECIMAL) {
            if (value === null || value === undefined) {
                return null;
            }
            return value?.toString();
        }

        return value;
    }


    setFieldValue(field: any, value: any) {
        if (value === undefined) return undefined;
        if (value === null) return null;

        if (field.type === AdminForthDataTypes.DATETIME) {
            if (value === "" || value === null) {
                return null;
            }
            return dayjs(value).isValid() ? dayjs(value).toDate() : null;
        }

        if (field.type === AdminForthDataTypes.INTEGER) {
            if (value === "" || value === null) {
                return null;
            }
            return Number.isFinite(value) ? Math.trunc(value) : null;
        }

        if (field.type === AdminForthDataTypes.FLOAT) {
            if (value === "" || value === null) {
                return null;
            }
            return Number.isFinite(value) ? value : null;
        }

        if (field.type === AdminForthDataTypes.DECIMAL) {
            if (value === "" || value === null) {
                return null;
            }
            return value.toString();
        }

        return value;
    }

    getFilterQuery(resource: AdminForthResource, filter: IAdminForthSingleFilter | IAdminForthAndOrFilter): any {
        // accept raw NoSQL filters for MongoDB
        if ((filter as IAdminForthSingleFilter).insecureRawNoSQL !== undefined) {
            return (filter as IAdminForthSingleFilter).insecureRawNoSQL;
        }

        // explicitly ignore raw SQL filters for MongoDB
        if ((filter as IAdminForthSingleFilter).insecureRawSQL !== undefined) {
            afLogger.warn(`⚠️  Ignoring insecureRawSQL filter for MongoDB:, ${(filter as IAdminForthSingleFilter).insecureRawSQL}`);
            return {};
        }

        if ((filter as IAdminForthSingleFilter).field) {
            // Field-to-field comparisons via $expr
            if ((filter as IAdminForthSingleFilter).rightField) {
                const left = `$${(filter as IAdminForthSingleFilter).field}`;
                const right = `$${(filter as IAdminForthSingleFilter).rightField}`;
                const op = (filter as IAdminForthSingleFilter).operator;
                const exprOpMap = {
                    [AdminForthFilterOperators.GT]: '$gt',
                    [AdminForthFilterOperators.GTE]: '$gte',
                    [AdminForthFilterOperators.LT]: '$lt',
                    [AdminForthFilterOperators.LTE]: '$lte',
                    [AdminForthFilterOperators.EQ]: '$eq',
                    [AdminForthFilterOperators.NE]: '$ne',
                } as const;
                const mongoExprOp = (exprOpMap as any)[op as any] as any;
                if (!mongoExprOp) {
                    // For unsupported ops with rightField, return empty condition
                    return {};
                }
                return { $expr: { [mongoExprOp]: [left, right] } };
            }
            const column = resource.dataSourceColumns.find((col) => col.name === (filter as IAdminForthSingleFilter).field);
            const filterValue = (filter as IAdminForthSingleFilter).value;
            if (column && [AdminForthDataTypes.INTEGER, AdminForthDataTypes.DECIMAL, AdminForthDataTypes.FLOAT].includes(column.type as any)) {
                // Handle array values for IN/NIN operators
                const convertedValue = Array.isArray(filterValue) 
                    ? filterValue.map(v => +v) 
                    : +filterValue;
                return { [(filter as IAdminForthSingleFilter).field as any]: ((this.OperatorsMap as any)[filter.operator as any] as any)(convertedValue) };
            }
            return { [(filter as IAdminForthSingleFilter).field as any]: ((this.OperatorsMap as any)[filter.operator as any] as any)(filterValue) };
        }

        // filter is a AndOr filter
        return ((this.OperatorsMap as any)[filter.operator as any] as any)((filter as IAdminForthAndOrFilter).subFilters
            // mongodb should ignore raw SQL, but allow raw NoSQL
            .filter((f) => (f as IAdminForthSingleFilter).insecureRawSQL === undefined)
            .map((f) => this.getFilterQuery(resource, f)));
    }

    async getAggregateWithOriginalTypes({ resource, filters, aggregations, groupBy }: {
        resource: AdminForthResource;
        filters: IAdminForthAndOrFilter;
        aggregations: { [alias: string]: IAggregationRule };
        groupBy?: IGroupByRule | IGroupByRule[];
    }): Promise<Array<{ group?: string, [key: string]: any }>> {

        const collection = this.client.db().collection(resource.table);

        const match = filters?.subFilters?.length ? this.getFilterQuery(resource, filters) : {};

        const groupByRules = this.normalizeGroupByRules(groupBy);
        let groupId: any = null;
        if (groupByRules.length) {
            groupId = {};
        }
        for (const [index, groupByRule] of groupByRules.entries()) {
            const alias = this.getGroupByResultAlias(groupByRule, index, groupByRules.length);
            if (groupByRule.type === 'field') {
                const g = groupByRule as IGroupByField;
                groupId[alias] = `$${g.field}`;
                continue;
            }
            const g = groupByRule as IGroupByDateTrunc;
            const tz = g.timezone ?? 'UTC';
            const dateTruncSpec: any = { 
                date: `$${g.field}`, 
                unit: g.truncation, 
                timezone: tz,
            };
            if (g.truncation === 'week') {
                dateTruncSpec.startOfWeek = 'Mon';
            }
            groupId[alias] = { $dateTrunc: dateTruncSpec };
        }

        const groupStage: Record<string, any> = {
            _id: groupId,
        };

        for (const [alias, rule] of Object.entries(aggregations)) {
            switch (rule.operation) {
                case 'count': groupStage[alias] = { $sum: 1 }; break;
                case 'count_distinct': groupStage[alias] = { $addToSet: `$${rule.field}` }; break;
                case 'sum': groupStage[alias] = { $sum: { $toDouble: `$${rule.field}` } }; break;
                case 'avg': groupStage[alias] = { $avg: { $toDouble: `$${rule.field}` } }; break;
                case 'min': groupStage[alias] = { $min: { $toDouble: `$${rule.field}` } }; break;
                case 'max': groupStage[alias] = { $max: { $toDouble: `$${rule.field}` } }; break;
                case 'median': groupStage[alias] = { $push: { $toDouble: `$${rule.field}` } }; break;
            }
        }

        const pipeline: any[] = [];

        if (Object.keys(match).length) {
            pipeline.push({ $match: match });
        }

        pipeline.push({ $group: groupStage });

        
        pipeline.push({
            $project: {
                _id: 0,
                ...Object.fromEntries(groupByRules.map((groupByRule, index) => {
                    const alias = this.getGroupByResultAlias(groupByRule, index, groupByRules.length);
                    return [alias, groupByRule.type === 'date_trunc' ? {
                        $cond: {
                            if: { $eq: [{ $type: `$_id.${alias}` }, "date"] },
                            then: {
                                $dateToString: {
                                    format: "%Y-%m-%d",
                                    date: `$_id.${alias}`,
                                    timezone: (groupByRule as IGroupByDateTrunc).timezone ?? 'UTC'
                                }
                            },
                            else: `$_id.${alias}`
                        }
                    } : `$_id.${alias}`];
                })),
                ...Object.fromEntries(
                    Object.keys(groupStage)
                        .filter(k => k !== '_id')
                        .map(k => [k, aggregations[k]?.operation === 'count_distinct' ? { $size: `$${k}` } : `$${k}`])
                ),
            },
        });

        const calculateMedian = (arr: any[]) => {
            if (!Array.isArray(arr) || arr.length === 0) return null;
            const sorted = [...arr].sort((a, b) => a - b);
            const mid = Math.floor(sorted.length / 2);
            return sorted.length % 2 === 0 
                ? (sorted[mid - 1] + sorted[mid]) / 2 
                : sorted[mid];
        };

        const result = await collection.aggregate(pipeline).toArray();

        const medianAliases = Object.keys(aggregations).filter(
            alias => aggregations[alias].operation === 'median'
        );

        return result.map((row: any) => {
            medianAliases.forEach(alias => {
                row[alias] = calculateMedian(row[alias]);
            });
            return row;
        });
    }
    
    async getDataWithOriginalTypes({ resource, limit, offset, sort, filters, columns }:
        { 
            resource: AdminForthResource, 
            limit: number, 
            offset: number, 
            sort: { field: string, direction: AdminForthSortDirections }[], 
            filters: IAdminForthAndOrFilter,
            columns?: Array<{ name: string }>,
        }
    ): Promise<any[]> {


        if (sort.some(s => !checkIfFieldIsInsideResourceColumns(s.field, resource))) {
            throw new Error(`Invalid sort field: ${sort.find(s => !checkIfFieldIsInsideResourceColumns(s.field, resource))?.field}`);
        }
        // const columns = resource.dataSourceColumns.filter(c=> !c.virtual).map((col) => col.name).join(', ');
        const tableName = resource.table;


        const collection = this.client.db().collection(tableName);
        const query = filters.subFilters.length ? this.getFilterQuery(resource, filters) : {};

        const sortArray: any[] = sort.map((s) => {
            return [s.field, this.SortDirectionsMap[s.direction]];
        });

        const projection = columns
            ? Object.fromEntries(columns.map((col) => [col.name, 1]))
            : undefined;

        const result = await collection.find(query, projection ? { projection } : undefined)
            .sort(sortArray)
            .skip(offset)
            .limit(limit)
            .toArray();

        return result
    }

    async getCount({ resource, filters }: { 
        resource: AdminForthResource,
        filters: IAdminForthAndOrFilter,
    }): Promise<number> {
        let normalizedFilters = filters;

        if (filters) {
            // validate and normalize in case this method is called from dataAPI
            const filterValidation = this.validateAndNormalizeFilters(filters, resource);
            if (!filterValidation.ok) {
                throw new Error(filterValidation.error);
            }
            normalizedFilters = filterValidation.normalizedFilters as IAdminForthAndOrFilter;
        }
        const collection = this.client.db().collection(resource.table);
        const query = normalizedFilters.subFilters.length ? this.getFilterQuery(resource, normalizedFilters) : {};
        return await collection.countDocuments(query);
    }

    async getMinMaxForColumnsWithOriginalTypes({ resource, columns }: any) {
        const tableName = resource.table;
        const collection = this.client.db().collection(tableName);
        const result: Record<string, { min: any; max: any }> = {};

        for (const column of columns) {
            const [doc] = await collection
            .aggregate([
                { $group: { _id: null, min: { $min: `$${column.name}` }, max: { $max: `$${column.name}` } } },
                { $project: { _id: 0, min: 1, max: 1 } },
            ])
            .toArray();

            result[column.name] = {
            min: normalizeMongoValue(doc?.min),
            max: normalizeMongoValue(doc?.max),
            };
        }
        return result;
    }

    async createRecordOriginalValues({ resource, record }: any): Promise<string> {
        const tableName = resource.table;
        const collection = this.client.db().collection(tableName);
        const columns = Object.keys(record);
        const newRecord: any = {};
        for (const colName of columns) {
            newRecord[colName] = record[colName];
        }
        const ret = await collection.insertOne(newRecord);
        return ret.insertedId;
    }

    async updateRecordOriginalValues({ resource, recordId, newValues }: any) {
        const collection = this.client.db().collection(resource.table);
        await collection.updateOne({ [this.getPrimaryKey(resource)]: recordId }, { $set: newValues });
    }

    async deleteRecord({ resource, recordId }: any): Promise<boolean> {
        const primaryKey = this.getPrimaryKey(resource);
        const collection = this.client.db().collection(resource.table);
        const res = await collection.deleteOne({ [primaryKey]: recordId });
        return res.deletedCount > 0;
    }

    async deleteMany({ resource, recordIds }: { resource: AdminForthResource; recordIds: string[] }): Promise<number> {
        if (!recordIds || recordIds.length === 0) {
            return 0;
        }
        const collection = this.client.db().collection(resource.table);
        const res = await collection.deleteMany({[this.getPrimaryKey(resource)]: { $in: recordIds }});
        return res.deletedCount ?? 0;
    }

    async close() {
        await this.client.close()
    }
}

export default MongoConnector;