using adas_core.Application.Exceptions;
using adas_core.Application.Repositories.Interfaces;
using adas_core.Domain.Enums;
using adas_core.Domain.Models;
using adas_core.Domain.Models.AppSettings;
using adas_core.Domain.Models.Filter;
using adas_core.Domain.Models.GroupedObservations;
using adas_core.Domain.Models.MongoModels;
using adas_core.Domain.Utils;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MongoDB.Bson;
using MongoDB.Driver;
using System.Diagnostics;
using System.Text.RegularExpressions;
using MongoUtils = adas_core.Infrastructure.Utils.MongoUtils;
namespace adas_core.Infrastructure.Repositories;
///
/// Repository implementation for managing entities in MongoDB.
/// Provides specialized query, aggregation, retention, and expiration operations for patient clinical observations.
///
public class ObservationRepository : MongoRepository, IObservationRepository
{
private readonly ApiSettings _apiSettings;
private readonly ILogger _logger;
///
/// Initializes a new instance of the class.
///
/// The application API settings containing configuration values, including the collection name. Must not be .
/// The MongoDB database instance used to access the collection.
/// The logger used to record diagnostic and error information.
/// Thrown when is .
public ObservationRepository(
IOptions? apiSettings,
IMongoDatabase database,
ILogger logger) : base(database)
{
if (apiSettings != null)
{
_logger = logger;
_apiSettings = apiSettings.Value;
}
else
{
throw new ArgumentNullException(nameof(apiSettings));
}
} //For testing
///
/// Gets the name of the MongoDB collection used to store patient observations.
/// Falls back to the default "patients_observations" collection name when not configured in the API settings.
///
/// The collection name retrieved from the API settings, or "patients_observations" if not configured.
public override string GetCollectionName()
{
return _apiSettings.PatientsObservations ?? "patients_observations";
}
///
/// Asynchronously finds the most recent observations of a specific type (by coding system and code) for a patient.
/// Results are sorted by time in descending order and limited to entries.
///
/// The unique identifier of the patient.
/// The coding system used (for example, LOINC, SNOMED).
/// The code identifying the observation type within the coding system.
/// The maximum number of recent observations to return. Defaults to 2.
/// An containing the matching observations ordered from newest to oldest.
public async Task> FindLastObservations(ObjectId patientId, string codingSystem,
string code, int num = 2)
{
var filter = Builders.Filter.And(
Builders.Filter.Eq(ob => ob.PatientId, patientId),
Builders.Filter.Eq(ob => ob.CodingSystem, codingSystem),
Builders.Filter.Eq(ob => ob.Code, code)
);
var result = await Collection.FindAsync(
filter,
new FindOptions
{ Sort = Builders.Sort.Descending("time"), Limit = num }
);
return result.ToEnumerable();
}
///
/// Asynchronously finds the most recent observations of a specific coding system for a patient.
/// Results are sorted by time in descending order and limited to entries.
///
/// The unique identifier of the patient.
/// The coding system to filter observations by.
/// The maximum number of recent observations to return. Defaults to 10.
/// A containing the matching observations ordered from newest to oldest.
public async Task> FindLastObservationsByCodingSystem(ObjectId patientId,
string codingSystem, int num = 10)
{
var filter = Builders.Filter.And(
Builders.Filter.Eq(ob => ob.PatientId, patientId),
Builders.Filter.Eq(ob => ob.CodingSystem, codingSystem)
);
var result = await Collection.FindAsync(
filter,
new FindOptions
{ Sort = Builders.Sort.Descending("time"), Limit = num }
);
return await result.ToListAsync();
}
///
/// Asynchronously retrieves the most recent observations for a patient, optionally restricted to a list of observation names.
/// If is , all distinct observation names for the patient are used.
///
/// The unique identifier of the patient.
/// The maximum number of observations to return per observation name.
/// Optional list of observation names to restrict the query to. When , all distinct names are discovered.
/// A containing the aggregated latest observations across all matching names.
public async Task> AggregatedPatientLastObservations(ObjectId patientId, int num,
List? filterObservations = null)
{
filterObservations = await AggregatePatientObservations(patientId, filterObservations);
var results = new List();
foreach (var obs in filterObservations)
{
var builder = Builders.Filter;
var filter = builder.And(
builder.Eq(o => o.PatientId, patientId),
builder.Eq(o => o.Name, obs)
);
var cursor = await Collection.FindAsync(filter,
new FindOptions
{
Sort = Builders.Sort.Descending("time").Descending("id"),
Limit = num
});
results.AddRange(cursor.ToEnumerable());
}
return results;
}
///
/// Asynchronously retrieves the most recent observations for a patient that occurred on or before a given date,
/// optionally restricted to a list of observation names.
///
/// The unique identifier of the patient.
/// The maximum number of observations to return per observation name.
/// The inclusive upper bound (UTC) for the observation time field.
/// Optional list of observation names to restrict the query to. When , all distinct names are discovered.
/// A containing the matching observations.
public async Task> AggregatedPatientLastObservationsByLastDate(ObjectId patientId, int num,
DateTime lastDate, List? filterObservations = null)
{
filterObservations = await AggregatePatientObservations(patientId, filterObservations);
var results = new List();
foreach (var obs in filterObservations)
{
var filter = Builders.Filter.And(
Builders.Filter.Eq(o => o.PatientId, patientId),
Builders.Filter.Eq(o => o.Name, obs),
Builders.Filter.Lte(o => o.Time, lastDate)
);
var cursor = await Collection.FindAsync(
filter,
new FindOptions
{ Sort = Builders.Sort.Descending("time").Descending("_id"), Limit = num }
);
results.AddRange(await cursor.ToListAsync());
}
return results;
}
///
/// Asynchronously retrieves the most recent observations for a patient, with per-field limits and an optional "expired" flag filter.
/// When a specifies as , only expired observations are returned.
/// When is , all observations for the patient are returned.
///
/// The unique identifier of the patient.
/// Optional list of descriptors defining the names, limits, and expiration filter. When , all observations for the patient are returned.
///
/// A containing the matching observations.
/// Returns an empty list when an exception occurs while querying the database.
///
public async Task> AggregatedPatientLastObservationsByField(ObjectId patientId,
List? filterObservations)
{
try
{
var results = new List();
IAsyncCursor? cursor;
var builder = Builders.Filter;
if (filterObservations != null)
{
foreach (var obs in filterObservations)
{
FilterDefinition filter;
if (obs is { OnlyExpired: true, Name: not null })
filter = builder.And(
builder.Eq(o => o.PatientId, patientId),
builder.Eq(o => o.Name, obs.Name),
builder.Eq(o => o.Expired, obs.OnlyExpired)
);
else
filter = builder.And(
builder.Eq(o => o.PatientId, patientId),
builder.Eq(o => o.Name, obs.Name)
);
cursor = await Collection.FindAsync(
filter,
new FindOptions
{
Sort = Builders.Sort.Descending("time").Descending("id"),
Limit = obs.Last
});
results.AddRange(cursor.ToEnumerable());
}
}
else
{
var filter = builder.Eq(o => o.PatientId, patientId);
cursor = await Collection.FindAsync(
filter,
new FindOptions
{ Sort = Builders.Sort.Descending("time").Descending("id") });
results.AddRange(cursor.ToEnumerable());
}
return results;
}
catch (Exception ex)
{
_logger.LogError("Error aggregated patient last observations by field {exMessage}", ex.Message);
return [];
}
}
///
/// Asynchronously executes an aggregation pipeline that groups a patient's observations into time buckets
/// (second / minute / hour / day / shift / times) and computes per-bucket results such as first, last, min, max, sum, average, or count.
/// Supports a "since last observation" mode and a configurable look-back window based on the regularity.
///
/// The unique identifier of the patient.
/// A descriptor containing the observation name(s), regularity, look-back window, and the result computations to perform.
///
/// A with one document per time bucket.
/// Returns an empty list when an exception occurs while executing the pipeline.
///
public async Task> AggregatedPatientGroupedObservations(ObjectId patientId,
GroupedField groupedField)
{
try
{
DateTime? lastObsTime = null;
if (groupedField.Since == GroupedObservationEnum.Since.Last && !string.IsNullOrEmpty(groupedField.Name))
{
var lastObsList =
await AggregatedPatientLastObservations(patientId, 1, [groupedField.Name]);
var lastObs = lastObsList.FirstOrDefault();
if (lastObs != null)
{
lastObsTime = lastObs.Time;
lastObsTime = DateTime.SpecifyKind(lastObsTime.Value, DateTimeKind.Utc);
}
}
var fields = new BsonArray();
var match = new BsonDocument { { "patientid", patientId } };
if (groupedField.GetNames().Count > 1)
{
groupedField.GetNames().ForEach(name => { fields.Add(new BsonDocument { { "name", name } }); });
match.Add("$or", fields);
}
else
{
match.Add("name", groupedField.Name);
}
var projectDate = new BsonDocument
{
{ "y", new BsonDocument { { "$year", "$time" } } },
{ "M", new BsonDocument { { "$month", "$time" } } },
{ "d", new BsonDocument { { "$dayOfMonth", "$time" } } }
};
var projectTimeFromParts = new BsonDocument
{
{ "year", "$_id.year" },
{ "month", "$_id.month" },
{ "day", "$_id.day" }
};
var idDocument = new BsonDocument { { "year", "$y" }, { "month", "$M" }, { "day", "$d" } };
var fromDate = DateTime.MinValue;
if (groupedField.Regularity == GroupedObservationEnum.Regularity.Day)
{
if (groupedField.Since == GroupedObservationEnum.Since.Last && lastObsTime.HasValue)
fromDate = lastObsTime.Value.AddDays(-1 * groupedField.Max);
else
fromDate = DateTime.UtcNow.AddDays(-1 * groupedField.Max);
}
if (groupedField.Regularity is GroupedObservationEnum.Regularity.Hour
or GroupedObservationEnum.Regularity.Minute or GroupedObservationEnum.Regularity.Second
or GroupedObservationEnum.Regularity.Times or GroupedObservationEnum.Regularity.Shift)
{
projectDate.Add("h", new BsonDocument { { "$hour", "$time" } });
idDocument.Add("hour", "$h");
projectTimeFromParts.Add("hour", "$_id.hour");
//fromDate = DateTime.UtcNow.AddHours(-1 * groupedField.max);
if (groupedField.Since == GroupedObservationEnum.Since.Last && lastObsTime.HasValue)
fromDate = new DateTime(lastObsTime.Value.Year, lastObsTime.Value.Month, lastObsTime.Value.Day,
lastObsTime.Value.AddHours(1).Hour, 00, 00).AddHours(-1 * groupedField.Max);
else
fromDate = new DateTime(DateTime.UtcNow.Year, DateTime.UtcNow.Month, DateTime.UtcNow.Day,
DateTime.UtcNow.Hour, 00, 00).AddHours(-1 * groupedField.Max);
}
if (groupedField.Regularity == GroupedObservationEnum.Regularity.Minute ||
groupedField.Regularity == GroupedObservationEnum.Regularity.Second ||
groupedField.Regularity == GroupedObservationEnum.Regularity.Times
//No queremos los minutos cuando pedimos por turno, en principio. Revisar si en algún caso necesitamos los minutos, quitado de momento por problemas
// a la hora de devolver el last.
//|| groupedField.regularity == Regularity.Shift
)
{
projectDate.Add("m", new BsonDocument { { "$minute", "$time" } });
idDocument.Add("minute", "$m");
projectTimeFromParts.Add("minute", "$_id.minute");
if (groupedField.Regularity != GroupedObservationEnum.Regularity.Shift)
{
if (groupedField.Since == GroupedObservationEnum.Since.Last && lastObsTime.HasValue)
fromDate = lastObsTime.Value.AddMinutes(-1 * groupedField.Max);
else
fromDate = DateTime.UtcNow.AddMinutes(-1 * groupedField.Max);
}
}
if (groupedField.Regularity == GroupedObservationEnum.Regularity.Second ||
groupedField.Regularity == GroupedObservationEnum.Regularity.Times)
{
projectDate.Add("s", new BsonDocument { { "$second", "$time" } });
idDocument.Add("second", "$s");
projectTimeFromParts.Add("second", "$_id.second");
if (groupedField.Since == GroupedObservationEnum.Since.Last && lastObsTime.HasValue)
fromDate = lastObsTime.Value.AddSeconds(-1 * groupedField.Max);
else
fromDate = DateTime.UtcNow.AddSeconds(-1 * groupedField.Max);
}
if (groupedField.Regularity == GroupedObservationEnum.Regularity.Second ||
groupedField.Regularity == GroupedObservationEnum.Regularity.Times)
{
projectDate.Add("ms", new BsonDocument { { "$millisecond", "$time" } });
idDocument.Add("millisecond ", "$ms");
projectTimeFromParts.Add("millisecond", "$_id.millisecond");
}
projectDate.Add("name", "$name");
projectDate.Add("min", "$min");
projectDate.Add("max", "$max");
projectDate.Add("time", "$time");
projectDate.Add("value", "$value");
idDocument.Add("name", "$name");
//No queremos meter el mínimo y el maximo en la agrupación.
//idDocument.Add("min", "$min");
//idDocument.Add("max", "$max");
fromDate = DateTime.SpecifyKind(fromDate, DateTimeKind.Utc);
if (groupedField.Max > 0)
{
//Add filter lte for ICCA future observations like hour balance.
if (groupedField.Since == GroupedObservationEnum.Since.Last && lastObsTime.HasValue)
match.Add("time", new BsonDocument("$gte", fromDate).Add("$lte", lastObsTime.Value));
else
match.Add("time", new BsonDocument("$gte", fromDate).Add("$lte", DateTime.UtcNow));
}
var group = new BsonDocument
{
{ "_id", idDocument }
};
if (groupedField.Result.Count == 0 || groupedField.Result.Contains(GroupedObservationEnum.Result.First))
group.Add("first",
new BsonDocument
{
{
"$first",
new BsonDocument
{ { "time", "$time" }, { "value", "$value" }, { "min", "$min" }, { "max", "$max" } }
}
});
if (groupedField.Result.Contains(GroupedObservationEnum.Result.Last))
group.Add("last",
new BsonDocument
{
{
"$last",
new BsonDocument
{ { "time", "$time" }, { "value", "$value" }, { "min", "$min" }, { "max", "$max" } }
}
});
if (groupedField.Result.Contains(GroupedObservationEnum.Result.LastFilled))
group.Add("lastfilled",
new BsonDocument
{
{
"$last",
new BsonDocument
{ { "time", "$time" }, { "value", "$value" }, { "min", "$min" }, { "max", "$max" } }
}
});
if (groupedField.Result.Contains(GroupedObservationEnum.Result.Min))
group.Add("min",
new BsonDocument
{
{
"$min",
new BsonDocument
{ { "value", "$value" }, { "time", "$time" }, { "min", "$min" }, { "max", "$max" } }
}
});
if (groupedField.Result.Contains(GroupedObservationEnum.Result.Max))
group.Add("max",
new BsonDocument
{
{
"$max",
new BsonDocument
{
{ "value", "$value" }, { "time", "$time" }, { "min", "$min" }, { "max", "$max" }
}
}
});
if (groupedField.Result.Contains(GroupedObservationEnum.Result.Sum))
group.Add("sum", new BsonDocument { { "$sum", "$value" } });
if (groupedField.Result.Contains(GroupedObservationEnum.Result.Average))
group.Add("average", new BsonDocument { { "$avg", "$value" } });
if (groupedField.Result.Contains(GroupedObservationEnum.Result.Count))
group.Add("count", new BsonDocument { { "$sum", 1 } });
if (groupedField.Result.Contains(GroupedObservationEnum.Result.HalfHour))
group.Add("all",
new BsonDocument
{
{
"$push",
new BsonDocument
{
{ "value", "$value" }, { "time", "$time" }, { "min", "$min" }, { "max", "$max" }
}
}
});
//group.Add("halfHour", new BsonDocument { { "$unset", new BsonDocument { { "time", "$time" }, { "value", "$value" }, { "min", "$min" }, { "max", "$max" } } } });
var pipeline = new List
{
new()
{
{
"$match", match
}
},
new()
{
{
"$sort", new BsonDocument { { "time", 1 } }
}
},
new()
{
{
"$project", projectDate
}
},
new()
{
{
"$group", group
}
},
new()
{
{
"$addFields",
new BsonDocument
{
{ "time", new BsonDocument { { "$dateFromParts", projectTimeFromParts } } },
{ "isFilled", false }
}
}
},
new()
{
{
"$sort", new BsonDocument { { "time", 1 } }
}
}
};
if (groupedField is { Regularity: GroupedObservationEnum.Regularity.Times, Max: > 0 })
pipeline.Add(new BsonDocument("$limit", groupedField.Max));
//System.Diagnostics.Debug.WriteLine("AggregatedPatientGroupedObservations query: \n" + pipeline.ToJson());
_logger.LogDebug("Executing AggregatedPatientGroupedObservations query: {pipeline}:", pipeline.ToJson());
var resultsCursor =
await Collection.AggregateAsync(pipeline, new AggregateOptions { AllowDiskUse = true });
var results = resultsCursor.ToList();
return results;
}
catch (Exception ex)
{
_logger.LogError("Error agregated patient grouped observations {exMessage}", ex.Message);
return [];
}
}
///
/// Asynchronously inserts a single patient observation, automatically retrying with a new
/// when a MongoDB duplicate-key error is encountered. The retry strategy is bounded by an internal maximum.
///
/// The to insert. If a duplicate-key error occurs, a new identifier is generated and the insert is retried.
/// A representing the asynchronous insert operation.
/// Rethrown after the maximum number of retries has been reached when a duplicate-key error keeps occurring.
/// Rethrown when an unexpected error occurs during the insert operation.
public new async Task InsertOneAsync(PatientObservation patientObservation)
{
const int maxRetries = 2; // Número máximo de reintentos
var retryCount = 0;
while (retryCount < maxRetries)
try
{
await Collection.InsertOneAsync(patientObservation);
return;
}
catch (MongoWriteException ex) when (ex.WriteError.Category == ServerErrorCategory.DuplicateKey)
{
retryCount++;
_logger.LogWarning(
"Duplicate key error encountered. Retrying with new ObjectId. Attempt {attempt} of {maxRetries}",
retryCount, maxRetries);
if (retryCount < maxRetries) continue;
_logger.LogError(
"Maximum retry attempts reached. Could not insert document due to duplicate key error.");
throw; // Relanzar la excepción después de alcanzar el número máximo de reintentos
}
catch (Exception ex)
{
_logger.LogError("Error inserting patient observation: {exMessage}", ex.Message);
throw;
}
}
///
/// Asynchronously deletes all observations associated with the specified patient.
///
/// The of the patient whose observations should be removed.
/// A representing the asynchronous delete operation.
public async Task DeleteByPatientId(ObjectId id)
{
var filter = Builders.Filter.Eq(po => po.PatientId, id);
await Collection.DeleteManyAsync(filter);
}
///
/// Asynchronously returns a cursor over all observations for a given patient, using a server-side batch size of 100.
///
/// The unique identifier of the patient.
///
/// An that can be enumerated to retrieve the patient's observations.
///
public async Task> FindByPatientIdAsync(ObjectId patientId)
{
var filter = Builders.Filter.Eq(ob => ob.PatientId, patientId);
return await Collection.FindAsync(filter, new FindOptions { BatchSize = 100 });
}
///
/// Asynchronously returns a cursor over all observations for a patient that match a given coding system and observation name.
///
/// The unique identifier of the patient.
/// The coding system to filter by.
/// The observation name to filter by.
///
/// An containing the matching observations,
/// retrieved with a server-side batch size of 100.
///
public async Task> FindByPatientIdAndCodingSystemAsync(ObjectId patientId,
string codingSystem, string name)
{
var filter =
Builders.Filter.And(
Builders.Filter.Eq(ob => ob.PatientId, patientId),
Builders.Filter.Eq(ob => ob.CodingSystem, codingSystem),
Builders.Filter.Eq(ob => ob.Name, name)
);
return await Collection.FindAsync(filter, new FindOptions { BatchSize = 100 });
}
///
/// Asynchronously deletes a single observation by its identifier. This method hides the base
/// DeleteAsync(ObjectId) defined on because the base returns the deleted document,
/// while this implementation is fire-and-forget.
///
/// The of the observation to delete.
/// A representing the asynchronous delete operation.
public new async Task DeleteAsync(ObjectId id)
{
var filter = Builders.Filter.Eq(obs => obs.Id, id);
await Collection.DeleteOneAsync(filter);
}
///
/// Deletes all observations with the specified name that are older than the configured number of days.
/// Returns the documents that were deleted for downstream processing.
///
/// The observation name to filter by.
/// The retention window expressed in days. Observations older than UtcNow - retentionPolicyValue days are removed.
/// A containing the documents that were deleted.
public async Task> DeleteOlderDaysAsync(string name, int retentionPolicyValue)
{
var dateLimit = DateTime.UtcNow.AddDays(-1 * retentionPolicyValue);
var filter = Builders.Filter.And(
Builders.Filter.Eq(obs => obs.Name, name),
Builders.Filter.Lt(obs => obs.Time, dateLimit)
);
var documentsToDelete = await Collection.Find(filter).ToListAsync();
await Collection.DeleteManyAsync(filter);
return documentsToDelete;
}
///
/// Deletes all observations with the specified name that are older than the configured number of seconds.
/// Returns the documents that were deleted for downstream processing.
///
/// The observation name to filter by.
/// The retention window expressed in seconds. Observations older than UtcNow - retentionPolicyValue seconds are removed.
/// A containing the documents that were deleted.
public async Task> DeleteOlderSecondsAsync(string name, int retentionPolicyValue)
{
var dateLimit = DateTime.UtcNow.AddSeconds(-1 * retentionPolicyValue);
var filter = Builders.Filter.And(
Builders.Filter.Eq(obs => obs.Name, name),
Builders.Filter.Lt(obs => obs.Time, dateLimit)
);
var documentsToDelete = await Collection.Find(filter).ToListAsync();
await Collection.DeleteManyAsync(filter);
return documentsToDelete;
}
///
/// Keeps only the most recent observations for the specified name,
/// deleting all older ones. Returns the documents that were deleted for downstream processing.
///
/// The observation name to apply the retention policy to.
/// The maximum number of observations to retain. The remainder are deleted.
/// A containing the documents that were deleted. Returns an empty list when nothing had to be deleted.
public async Task> DeleteOlderNumberAsync(string name, int retentionPolicyValue)
{
var builder = Builders.Filter;
var idsToDelete = await Collection.Find(obs => obs.Name == name)
.Project(obs => obs.Id)
.Sort("{time: -1}")
.Skip(retentionPolicyValue)
.ToListAsync();
List documentsToDelete = [];
if (idsToDelete == null || !idsToDelete.Any()) return documentsToDelete;
var idFilter = builder.In("_id", idsToDelete);
documentsToDelete = await Collection.Find(idFilter).ToListAsync();
await Collection.DeleteManyAsync(idFilter);
return documentsToDelete;
}
///
/// Asynchronously checks whether an observation exists for a given patient with the specified SystemId.
/// Only the document identifier is projected, making the query lightweight.
///
/// The unique identifier of the patient.
/// The external system identifier to look up.
/// if a matching observation exists; otherwise, .
public async Task ExistBySystemId(ObjectId patientid, string systemId)
{
return await Collection.Find(Builders.Filter.And(
Builders.Filter.Eq(obs => obs.PatientId, patientid),
Builders.Filter.Eq(obs => obs.SystemId, systemId)
))
.Project("{ _id: true }")
.AnyAsync();
}
///
/// Asynchronously finds the most recent observation for a patient with the given name that occurred strictly before the specified date.
///
/// The unique identifier of the patient.
/// The observation name to search for. Can be .
/// The upper-bound (exclusive) observation time.
///
/// A representing the asynchronous operation.
/// The task result contains the most recent matching observation, or if no observation matches.
///
public async Task FindLastObservationBeforeDate(ObjectId patientId, string? name,
DateTime date)
{
var builder = Builders.Filter;
var filter = builder.And(
builder.Eq(obs => obs.PatientId, patientId),
builder.Eq(obs => obs.Name, name),
builder.Lt(obs => obs.Time, date)
);
var result = await Collection.FindAsync(filter,
new FindOptions
{ Sort = Builders.Sort.Descending("time").Descending("_id") });
return await result.FirstOrDefaultAsync();
}
///
/// Asynchronously retrieves all observations for a patient with a given name whose time exactly matches the provided value.
///
/// The unique identifier of the patient.
/// The observation name to match. Can be .
/// The exact time value to match.
///
/// A representing the asynchronous operation.
/// The task result contains a list of matching observations, which may be empty.
///
public async Task?> FindAnyWithSameDate(ObjectId patientId, string? name, DateTime date)
{
var builder = Builders.Filter;
var filter = builder.Eq(obs => obs.PatientId, patientId)
& builder.Eq(obs => obs.Name, name)
& builder.Eq(obs => obs.Time, date);
var result = await Collection.FindAsync(filter);
return result.ToList();
}
///
/// Asynchronously retrieves all observations for a patient whose time is strictly before the specified date.
///
/// The unique identifier of the patient.
/// The upper-bound (exclusive) observation time.
///
/// A containing the matching observations, which may be empty.
///
public async Task> FindAnyBeforeDate(ObjectId patientId, DateTime date)
{
var filterBuilder = Builders.Filter;
var patientIdFilter = filterBuilder.Eq(obs => obs.PatientId, patientId);
var dateFilter = filterBuilder.Lt(obs => obs.Time, date);
var filter = filterBuilder.And(patientIdFilter, dateFilter);
var result = await Collection.FindAsync(filter);
return result.ToList();
}
///
/// Asynchronously retrieves, for a given patient and observation name, the latest occurrence of each distinct value.
/// Optionally restricts the result to observations not older than seconds.
///
/// The unique identifier of the patient.
/// The observation name to search for.
///
/// Optional expiration window expressed in seconds. When provided, only observations newer than
/// UtcNow - expires seconds are considered.
///
///
/// A containing the most recent observation for each distinct value.
///
public async Task> FindLatestUniqueValuesByName(ObjectId patientId, string name,
int? expires)
{
// Crear el filtro base
var matchFilter = new BsonDocument
{
{ "patientid", patientId },
{ "name", name }
};
// Agregar filtro de expiración si expires no es null
if (expires.HasValue)
{
var expirationDate = DateTime.UtcNow.AddSeconds(-expires.Value);
matchFilter.Add("time", new BsonDocument("$gte", expirationDate));
}
var pipeline = new[]
{
// Filtrar por PatientId, name y (opcionalmente) tiempo no expirado
new BsonDocument("$match", matchFilter),
// Ordenar por tiempo en orden descendente
new BsonDocument("$sort", new BsonDocument("time", -1)),
// Agrupar por el campo `value` (valores únicos)
new BsonDocument("$group", new BsonDocument
{
{ "_id", "$value" },
{ "LatestObservation", new BsonDocument("$first", "$$ROOT") }
}),
// Proyectar solo los campos originales
new BsonDocument("$replaceRoot", new BsonDocument("newRoot", "$LatestObservation"))
};
var result = await Collection.AggregateAsync(pipeline);
return result.ToList();
}
///
/// Asynchronously retrieves the currently active intravenous-line observations for a patient,
/// grouping by the Location and Type of the .
/// Within each group, only the most recent observation (ordered by time and id) is returned.
///
/// The unique identifier of the patient.
///
/// A containing one observation per unique (Location, Type) combination.
///
public async Task> AggregatedPatientActiveIntravenousLinesObservations(ObjectId patientId)
{
var results = new List();
var cursor = await Collection.FindAsync(
o => o.PatientId == patientId && o.Name == "IntravenousLinesObs",
new FindOptions
{ Sort = Builders.Sort.Descending("time").Descending("id") });
results.AddRange(cursor.ToEnumerable());
//Group by tpye and location, cant be in same location two of same type.
return results
.GroupBy(obs => new
{
((PatientIntravenousLinesValue)obs.Value).Location,
((PatientIntravenousLinesValue)obs.Value).Type
})
.Select(g =>
g.OrderBy(t => t.Time).ThenBy(t => t.Id).LastOrDefault())
.ToList();
}
///
/// Asynchronously retrieves, for every patient with observations, the timestamp of the most recent observation.
/// Implemented as a server-side aggregation that groups by patientid and selects the latest time value.
///
///
/// A mapping each patient's to the UTC timestamp of their latest observation.
///
public async Task> FindAllLastPatientObservationTime()
{
var group = new BsonDocument
{
{ "_id", "$patientid" },
{ "time", new BsonDocument { { "$last", "$time" } } }
};
var pipeline = new List
{
new()
{
{
"$sort", new BsonDocument { { "time", 1 } }
}
},
new()
{
{
"$group", group
}
}
};
var cursor = await Collection.AggregateAsync(pipeline,
new AggregateOptions { AllowDiskUse = true, BatchSize = 10 });
var result = new Dictionary();
while (await cursor.MoveNextAsync())
cursor.Current
.Where(obs => obs.GetValue("_id").BsonType != BsonType.Null)
.ToList()
.ForEach(obs =>
{
if (obs.Get("_id") != null)
result.Add(obs.Get("_id")?.AsObjectId ?? new ObjectId(),
obs.Get("time")?.ToUniversalTime() ?? DateTime.MinValue);
}
);
return result;
}
///
/// Asynchronously finds an observation by its unique identifier.
///
/// The of the observation to retrieve.
///
/// A representing the asynchronous operation.
/// The task result contains the if found; otherwise, .
///
public async Task FindById(ObjectId id)
{
var filter = Builders.Filter.Eq(o => o.Id, id);
var result = await Collection.FindAsync(filter);
return await result.FirstOrDefaultAsync();
}
///
/// Asynchronously retrieves all observations for a patient by the patient's identifier.
///
/// The of the patient whose observations should be retrieved.
///
/// A representing the asynchronous operation.
/// The task result contains a list of observations, which may be empty.
///
public async Task?> FindByPatientId(ObjectId id)
{
var filter = Builders.Filter.Eq(o => o.PatientId, id);
var result = await Collection.FindAsync(filter);
return await result.ToListAsync();
}
///
/// Asynchronously updates a single observation by replacing its document with the provided instance.
///
/// The whose identifies the document to update.
/// A representing the asynchronous update operation.
public async Task Update(PatientObservation observation)
{
await UpdateOneAsync(observation.Id, observation);
}
///
/// Asynchronously updates all documents in the collection where the specified field equals ,
/// setting that field to the new . Thin wrapper around the protected helper on the base repository.
///
/// The name of the field to match and update.
/// The new value to assign.
/// The existing value to replace.
/// A representing the asynchronous update operation.
public async Task UpdateManyObjectId(string nameId, ObjectId id, ObjectId oldId)
{
await UpdateManyObjectIdAsync(nameId, id, oldId);
}
///
/// Asynchronously marks the supplied list of observations as expired by setting their Expired flag to .
///
/// The list of instances to mark as expired. The set of identifiers is used to build the update filter.
/// A representing the asynchronous update operation.
public async Task UpdateExpiredObservations(List expiredObservations)
{
var filter = Builders.Filter
.In(x => x.Id, expiredObservations.Select(c => c.Id));
//var filter = Builders.Filter.In("Id", expiredObservations);
var update = Builders.Update.Set("Expired", true);
_ = await Collection.UpdateManyAsync(filter, update);
}
///
/// Asynchronously applies the given update definition to a set of observations identified by their identifiers.
///
/// The observations whose identifiers form the target set of the update.
/// The describing the changes to apply to each matching document.
/// A representing the asynchronous update operation.
public async Task UpdateMany(IEnumerable patientObservations,
UpdateDefinition update)
{
var filter = Builders.Filter.In(x => x.Id, patientObservations.Select(x => x.Id));
await Collection.UpdateManyAsync(filter, update);
}
///
/// Asynchronously returns every observation stored in the collection.
///
///
/// An containing all observations.
///
public async Task> FindAll()
{
var result = await Collection.FindAsync(_ => true);
return result.ToEnumerable();
}
///
/// Asynchronously marks observations as expired when their time is older than the configured expiration
/// window defined by the supplied entries. Only observations that are not
/// already marked as expired are updated. Errors are logged and swallowed.
///
///
/// A list of entries describing the observation names and their expiration windows
/// (in minutes). Observations whose time is older than DateTime.Now - expires minutes are marked as expired.
///
/// A representing the asynchronous update operation.
public async Task ExpireExpiredObservations(
List configObservationsToExpire)
{
try
{
var builder = Builders.Filter;
var filterObs = builder.Empty;
configObservationsToExpire.ForEach(ob =>
{
var minsToAdd = ob.Expires.HasValue ? ob.Expires * -1 : 0;
if (filterObs == builder.Empty)
filterObs = builder.Where(o =>
o.Name == ob.Name && o.Time < DateTime.Now.AddMinutes(minsToAdd.Value));
else
filterObs |= builder.Where(o =>
o.Name == ob.Name && o.Time < DateTime.Now.AddMinutes(minsToAdd.Value));
});
var filterNotExpired = Builders.Filter.Where(o => !o.Expired);
var combineFilter = filterObs & filterNotExpired;
/*var documentSerializer = BsonSerializer.SerializerRegistry.GetSerializer();
var renderedFilter = combineFilter.Render(documentSerializer, BsonSerializer.SerializerRegistry);
Debug.WriteLine($"Result : {renderedFilter}");*/
var updateDefinition = Builders.Update.Set(o => o.Expired, true);
await Collection.UpdateManyAsync(combineFilter, updateDefinition);
}
catch (Exception ex)
{
_logger.LogError("exception expiring observations expired: {exMessage} ", ex);
}
}
///
/// Asynchronously retrieves observations that are not marked as expired, optionally restricted to a list of observation names.
/// Observations with a name are excluded from the result.
///
///
/// Optional list of observation names to match. When , all non-null named observations are considered
/// (still subject to the not-expired filter).
///
///
/// An containing the matching non-expired observations.
///
public async Task> FindNotExpired(List? filterObservations)
{
var filterBuilder = Builders.Filter;
var filters = new List>();
if (filterObservations != null)
{
filters.Add(filterBuilder.Not(filterBuilder.Eq(o => o.Name, null)));
filters.Add(filterBuilder.In(o => o.Name, filterObservations));
}
filters.Add(filterBuilder.Ne(o => o.Expired, true));
var filter = filterBuilder.And(filters);
var result = await Collection.FindAsync(filter);
return await result.ToListAsync();
}
///
/// Asynchronously finds observations whose name matches the given pattern (case-insensitive substring/regex),
/// optionally restricted to those with a time greater than .
///
/// The regex pattern to match against the observation name. Cannot be null, empty, whitespace, or longer than 100 characters.
/// Optional inclusive lower bound on the observation time.
///
/// An containing the matching observations.
///
///
/// Thrown when is null, empty, or whitespace,
/// or when its length exceeds 100 characters.
///
public async Task> FindByName(string name, DateTime? fromDate = null)
{
if (string.IsNullOrWhiteSpace(name))
throw new BadRequestException(HttpEnum.ErrorMessage.BadRequestInvalidFormat);
if (name.Length > 100)
throw new BadRequestException("Name too long");
var safeName = Regex.Escape(name);
var filterBuilder = Builders.Filter;
var nameFilter = filterBuilder.And(
filterBuilder.Ne(o => o.Name, null),
filterBuilder.Regex(o => o.Name, new BsonRegularExpression(safeName, "i"))
);
var dateFilter = fromDate.HasValue
? filterBuilder.Gt(o => o.Time, fromDate.Value)
: filterBuilder.Empty;
var filter = filterBuilder.And(nameFilter, dateFilter);
var result = await Collection.FindAsync(filter);
return await result.ToListAsync();
}
///
/// Builds a paginated, sorted, and filtered query over observations.
/// Results are sorted by time in descending order. When a
/// is provided, observations can be filtered by name list and patient identifier. A mandatory time window
/// (defaulting to DateTime.MinValue / DateTime.MaxValue) is always applied.
///
/// The containing pagination and filter criteria.
///
/// An instance that can be used to further refine and execute the query.
///
public IFindFluent GetPaginatedObservations(PaginationFilter filter)
{
// Crear variable con la clase que construye los filtros
var filterBuilder = Builders.Filter;
// Crear una lista de filtros
var filters = new List>();
// Ordenar los resultados por "time" en orden descendente
var sort = Builders.Sort.Descending("time");
if (filter.FilteredRequest != null)
{
var requestFilter = filter.FilteredRequest;
if (!requestFilter.Observations.IsNullOrEmpty())
filters.Add(filterBuilder.In(o => o.Name, requestFilter.Observations));
if (requestFilter.PatientId != null)
{
var parsed = ObjectId.TryParse(requestFilter.PatientId, out var patientObjectId);
if (parsed) filters.Add(filterBuilder.Eq(o => o.PatientId, patientObjectId));
}
}
// Añadir los filtros por defecto en caso de llegar la lista de filtros vacia, si no la consulta a base de datos falla
filters.Add(filterBuilder.Gt(p => p.Time, filter.FilteredRequest?.StartDate ?? DateTime.MinValue));
filters.Add(filterBuilder.Lt(p => p.Time, filter.FilteredRequest?.EndDate ?? DateTime.MaxValue));
// Combinar los filtros en una consulta compuesta con operador "$and"
var combinedFilter = Builders.Filter.And(filters);
return Collection
.Find(combinedFilter)
.Sort(sort);
}
///
/// Asynchronously retrieves the most recent observations for a patient with a given name,
/// optionally bounded to a recent time window and an explicit maximum count.
///
/// The unique identifier of the patient.
/// The observation name to filter by.
///
/// Optional look-back window in seconds. When provided, only observations whose time is greater than or equal to
/// UtcNow - endAfter are returned.
///
/// Optional maximum number of observations to return. means no limit.
///
/// An containing the matching observations ordered from newest to oldest.
///
public async Task> FindLastNotExpiredObservatonsByPatient(ObjectId patientId,
string name, int? endAfter = null, int? num = null)
{
var sort = Builders.Sort.Descending("time");
var filterBuilder = Builders.Filter;
if (endAfter.HasValue)
{
var now = DateTime.UtcNow;
var thresholdTime = now.AddSeconds(-endAfter.Value); // Calcular la fecha límite
var dateFilter =
filterBuilder.Gte(o => o.Time, thresholdTime); // con fecha mayor que la fecha límite calculada
filterBuilder.And(dateFilter);
}
var patientFilter = filterBuilder.Eq(o => o.PatientId, patientId);
var nameFilter = filterBuilder.Eq(o => o.Name, name);
var filters = filterBuilder.And(patientFilter, nameFilter);
var result = await Collection.FindAsync(
filters,
new FindOptions { Sort = sort, Limit = num }
);
return result.ToEnumerable();
}
///
/// Creates the indexes required by the observations collection to support the repository's query patterns.
/// Indexes are created in the background and are non-unique. If an error occurs, it is logged and rethrown.
///
/// A representing the asynchronous index creation operation.
/// Rethrown when an error occurs while creating the indexes.
public override async Task CreateIndexes()
{
try
{
var options = new CreateIndexOptions { Background = true, Unique = false };
var indexes = new List>
{
new("{ patientid: 1 }", options),
new("{ patientid: 1, name: 1, codingSystem: 1 }", options),
new("{ patientid: 1, name: 1 }", options),
new("{ name: 1, time: -1 }", options),
new("{ patientid: 1, systemId: 1 }", options),
new("{ name: 1 }", options),
new("{ patientid: 1, name: 1 , time: -1, id: -1}", options),
new("{ patientid: 1, name: 1 , time: 1}", options)
};
await MongoUtils.EnsureIndexes(Collection, indexes);
}
catch (Exception e)
{
_logger.LogError(
"error creating indexes for observation collection {eMessage} TRACE: {eStackTrace}", e.Message,
e.StackTrace);
throw;
}
}
///
/// Determines the list of observation names to use when running a per-name aggregation for a patient.
/// If is null or empty, the distinct observation names currently
/// stored for the patient are discovered via a server-side $group aggregation.
///
/// The unique identifier of the patient.
///
/// Optional list of observation names to use. When null or empty, the method discovers the patient's
/// distinct observation names automatically.
///
///
/// A containing the observation names to aggregate over.
/// Returns an empty list if no observations exist for the patient.
///
private async Task> AggregatePatientObservations(ObjectId patientId,
List? filterObservations = null)
{
var matchPatient = new BsonDocument
{
{ "patientid", patientId },
{ "name", new BsonDocument { { "$ne", BsonNull.Value } } }
};
if (filterObservations == null || !filterObservations.Any())
{
// GET DISTINCT OBSERVATIONS
var distinctObs = new BsonDocument
{
{
"$group", new BsonDocument
{
{ "_id", "1" },
{
"obs", new BsonDocument
{
{ "$addToSet", "$name" }
}
}
}
}
};
var distinctPipeline = new[]
{
new()
{
{
"$match", matchPatient
}
},
distinctObs
};
Debug.WriteLine("AggregatedPatientLastObservations distinct obs: \n" +
distinctPipeline.ToJson());
var resultList = await Collection.AggregateAsync(distinctPipeline,
new AggregateOptions { AllowDiskUse = true });
var result = resultList.ToList().FirstOrDefault();
if (result != null)
filterObservations = result.GetValue("obs").AsBsonArray.Select(it => it.AsString)
.ToList();
}
return filterObservations ?? [];
}
}