forked from ddrilling/AsbCloudServer
355 lines
14 KiB
C#
355 lines
14 KiB
C#
using AsbCloudApp.Data;
|
|
using AsbCloudApp.Services;
|
|
using AsbCloudDb.Model;
|
|
using AsbCloudInfrastructure.Services.Cache;
|
|
using Mapster;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading.Tasks;
|
|
using System;
|
|
using System.Threading;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using AsbCloudDb;
|
|
|
|
namespace AsbCloudInfrastructure.Services
|
|
{
|
|
public class TelemetryService : ITelemetryService
|
|
{
|
|
private readonly CacheTable<Telemetry> cacheTelemetry;
|
|
private readonly CacheTable<Well> cacheWells;//TODO: use wellService instead of this
|
|
private readonly IAsbCloudDbContext db;
|
|
private readonly ITelemetryTracker telemetryTracker;
|
|
private readonly ITimezoneService timezoneService;
|
|
|
|
public ITimezoneService TimeZoneService => timezoneService;
|
|
public ITelemetryTracker TelemetryTracker => telemetryTracker;
|
|
|
|
public TelemetryService(
|
|
IAsbCloudDbContext db,
|
|
ITelemetryTracker telemetryTracker,
|
|
ITimezoneService timezoneService,
|
|
CacheDb cacheDb)
|
|
{
|
|
cacheTelemetry = cacheDb.GetCachedTable<Telemetry>((AsbCloudDbContext)db);
|
|
cacheWells = cacheDb.GetCachedTable<Well>(
|
|
(AsbCloudDbContext)db,
|
|
$"{nameof(Well.Cluster)}.{nameof(Cluster.Deposit)}",
|
|
nameof(Well.Telemetry),
|
|
$"{nameof(Well.RelationCompaniesWells)}.{nameof(RelationCompanyWell.Company)}",
|
|
nameof(Well.WellType));
|
|
this.db = db;
|
|
this.telemetryTracker = telemetryTracker;
|
|
this.timezoneService = timezoneService;
|
|
}
|
|
|
|
public IEnumerable<TelemetryDto> GetTransmittingTelemetries()
|
|
{
|
|
var telemetryDtos = new List<TelemetryDto>();
|
|
var activeTelemetriesUids = telemetryTracker.GetTransmittingTelemetriesUids();
|
|
if (!activeTelemetriesUids.Any())
|
|
return telemetryDtos;
|
|
var telemetries = cacheTelemetry
|
|
.Where(t => activeTelemetriesUids.Contains(t.RemoteUid));
|
|
telemetryDtos = telemetries.Adapt<TelemetryDto>().ToList();
|
|
|
|
return telemetryDtos;
|
|
}
|
|
|
|
public void SaveRequestDate(string uid, DateTimeOffset remoteDate) =>
|
|
telemetryTracker.SaveRequestDate(uid, remoteDate);
|
|
|
|
public DateTime GetLastTelemetryDate(int idTelemetry, bool useUtc = false)
|
|
{
|
|
var lastTelemetryDate = DateTimeOffset.MinValue;
|
|
var telemetry = cacheTelemetry.FirstOrDefault(t => t.Id == idTelemetry);
|
|
|
|
if (telemetry is null)
|
|
throw new Exception($"Telemetry id:{idTelemetry} does not exist");
|
|
|
|
var uid = telemetry.RemoteUid;
|
|
var timzone = GetTimezone(idTelemetry);
|
|
|
|
lastTelemetryDate = telemetryTracker.GetLastTelemetryDateByUid(uid);
|
|
|
|
if (useUtc)
|
|
return lastTelemetryDate.UtcDateTime;
|
|
|
|
return lastTelemetryDate.ToRemoteDateTime(timzone.Hours);
|
|
}
|
|
|
|
public DatesRangeDto GetDatesRange(int idTelemetry)
|
|
{
|
|
var telemetry = cacheTelemetry.FirstOrDefault(t => t.Id == idTelemetry);
|
|
if (telemetry is null)
|
|
throw new Exception($"Telemetry id:{idTelemetry} does not exist");
|
|
|
|
var dto = TelemetryTracker.GetTelemetryDateRangeByUid(telemetry.RemoteUid);
|
|
if (dto is null)
|
|
throw new Exception($"Telemetry id:{idTelemetry} has no data");
|
|
|
|
var timezone = GetTimezone((int)idTelemetry);
|
|
dto.From = dto.From.ToTimeZoneOffsetHours(timezone.Hours);
|
|
dto.To = dto.To.ToTimeZoneOffsetHours(timezone.Hours);
|
|
|
|
return dto;
|
|
}
|
|
|
|
public int GetOrCreateTelemetryIdByUid(string uid)
|
|
=> GetOrCreateTelemetryByUid(uid).Id;
|
|
|
|
public int? GetIdWellByTelemetryUid(string uid)
|
|
=> GetWellByTelemetryUid(uid)?.Id;
|
|
|
|
public async Task UpdateInfoAsync(string uid, TelemetryInfoDto info,
|
|
CancellationToken token)
|
|
{
|
|
var telemetry = GetOrCreateTelemetryByUid(uid);
|
|
telemetry.Info = info.Adapt<TelemetryInfo>();
|
|
|
|
if (!string.IsNullOrEmpty(info.TimeZoneId) &&
|
|
telemetry.TimeZone?.IsOverride != true)
|
|
telemetry.TimeZone = new SimpleTimezone()
|
|
{
|
|
Hours = info.TimeZoneOffsetTotalHours,
|
|
TimeZoneId = info.TimeZoneId
|
|
};
|
|
|
|
await cacheTelemetry.UpsertAsync(telemetry, token)
|
|
.ConfigureAwait(false);
|
|
}
|
|
|
|
public SimpleTimezoneDto GetTimezone(int idTelemetry)
|
|
{
|
|
var telemetry = cacheTelemetry.FirstOrDefault(t => t.Id == idTelemetry);
|
|
|
|
if (telemetry is null)
|
|
throw new Exception($"Telemetry id: {idTelemetry} does not exist.");
|
|
|
|
if (telemetry.TimeZone is not null)
|
|
return telemetry.TimeZone.Adapt<SimpleTimezoneDto>();
|
|
|
|
if (telemetry.Info is not null)
|
|
{
|
|
telemetry.TimeZone = new SimpleTimezone
|
|
{
|
|
Hours = telemetry.Info.TimeZoneOffsetTotalHours,
|
|
IsOverride = false,
|
|
TimeZoneId = telemetry.Info.TimeZoneId,
|
|
};
|
|
cacheTelemetry.Upsert(telemetry);
|
|
return telemetry.TimeZone.Adapt<SimpleTimezoneDto>();
|
|
}
|
|
|
|
if (telemetry.Well?.Timezone is not null)
|
|
{
|
|
telemetry.TimeZone = telemetry.Well.Timezone;
|
|
cacheTelemetry.Upsert(telemetry);
|
|
return telemetry.TimeZone.Adapt<SimpleTimezoneDto>();
|
|
}
|
|
|
|
throw new Exception($"Telemetry id: {idTelemetry} can't find timezone.");
|
|
}
|
|
|
|
public async Task UpdateTimezoneAsync(string uid, SimpleTimezoneDto timeone,
|
|
CancellationToken token)
|
|
{
|
|
var telemetry = GetOrCreateTelemetryByUid(uid);
|
|
var newTelemetryTimeZone = timeone.Adapt<SimpleTimezone>();
|
|
if (newTelemetryTimeZone?.Equals(telemetry.TimeZone) == true)
|
|
return;
|
|
await cacheTelemetry.UpsertAsync(telemetry, token)
|
|
.ConfigureAwait(false);
|
|
}
|
|
|
|
public int? GetIdTelemetryByIdWell(int idWell)
|
|
{
|
|
var well = cacheWells.FirstOrDefault(w => w.Id == idWell);
|
|
return well?.IdTelemetry;
|
|
}
|
|
|
|
private Well GetWellByTelemetryUid(string uid)
|
|
{
|
|
var tele = cacheTelemetry.FirstOrDefault(t => t.RemoteUid == uid);
|
|
if (tele is null)
|
|
return null;
|
|
|
|
var well = cacheWells.FirstOrDefault(w => w?.IdTelemetry == tele.Id);
|
|
return well;
|
|
}
|
|
|
|
private Telemetry GetOrCreateTelemetryByUid(string uid)
|
|
=> cacheTelemetry.GetOrCreate(t => t.RemoteUid == uid, () => new Telemetry
|
|
{
|
|
RemoteUid = uid,
|
|
TimeZone = new SimpleTimezone
|
|
{
|
|
Hours = 5,
|
|
IsOverride = false,
|
|
TimeZoneId = "default",
|
|
}
|
|
});
|
|
|
|
public async Task<int> MergeAsync(int from, int to, CancellationToken token)
|
|
{
|
|
if (from == to)
|
|
return -2;
|
|
|
|
var stopwath = new System.Diagnostics.Stopwatch();
|
|
stopwath.Start();
|
|
|
|
var transaction = await db.Database.BeginTransactionAsync(token).ConfigureAwait(false);
|
|
try
|
|
{
|
|
var affected = 0;
|
|
|
|
var wellFrom = await db.Wells.FirstOrDefaultAsync(w => w.IdTelemetry == from, token)
|
|
.ConfigureAwait(false);
|
|
var wellTo = await db.Wells.FirstOrDefaultAsync(w => w.IdTelemetry == to, token)
|
|
.ConfigureAwait(false);
|
|
|
|
if (wellTo is not null && wellFrom is not null)
|
|
return -2;
|
|
|
|
if (wellTo is null && wellFrom is not null)
|
|
{
|
|
wellFrom.IdTelemetry = to;
|
|
affected += await db.SaveChangesAsync(token);
|
|
}
|
|
|
|
affected += await MergeEventsAndMessagesAndUsersAsync(from, to, token);
|
|
affected += await MergeDataAsync<TelemetryDataSaub>(from, to, token);
|
|
affected += await MergeDataAsync<TelemetryDataSpin>(from, to, token);
|
|
|
|
affected += await db.Database.ExecuteSqlRawAsync($"DELETE FROM t_telemetry_analysis WHERE id_telemetry = {from} OR id_telemetry = {to};", token)
|
|
.ConfigureAwait(false);
|
|
affected += await db.Database.ExecuteSqlRawAsync($"DELETE FROM t_telemetry WHERE id = {from};", token)
|
|
.ConfigureAwait(false);
|
|
await transaction.CommitAsync(token).ConfigureAwait(false);
|
|
|
|
stopwath.Stop();
|
|
Console.WriteLine($"Successfully committed in {1d * stopwath.ElapsedMilliseconds / 1000d: #0.00} sec. Affected {affected} rows.");
|
|
return affected;
|
|
}
|
|
catch(Exception ex)
|
|
{
|
|
System.Diagnostics.Trace.WriteLine($"Merge() Fail. Rollback. Reason is:{ex.Message}");
|
|
await transaction.RollbackAsync(token).ConfigureAwait(false);
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
private async Task<int> MergeEventsAndMessagesAndUsersAsync(int from, int to, CancellationToken token)
|
|
{
|
|
var messagesFrom = await db.TelemetryMessages
|
|
.Where(d => d.IdTelemetry == from)
|
|
.ToListAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
var usersFromQuery = db.TelemetryUsers
|
|
.Where(d => d.IdTelemetry == from);
|
|
var usersFrom = await usersFromQuery
|
|
.ToListAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
var usersTo = await db.TelemetryUsers
|
|
.Where(d => d.IdTelemetry == to)
|
|
.AsNoTracking()
|
|
.ToListAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
var usersToNextId = usersTo.Max(u => u.IdUser) + 100;
|
|
|
|
messagesFrom.ForEach(m => m.IdTelemetry = to);
|
|
|
|
foreach (var userFrom in usersFrom)
|
|
{
|
|
var userTo = usersTo
|
|
.FirstOrDefault(u=>u.IdUser == userFrom.IdUser);
|
|
|
|
if (userTo is null ||
|
|
userTo.Name != userFrom.Name ||
|
|
userTo.Surname != userFrom.Surname ||
|
|
userTo.Patronymic != userFrom.Patronymic)
|
|
{
|
|
messagesFrom
|
|
.Where(m => m.IdTelemetryUser == userFrom.IdUser)
|
|
.ToList()
|
|
.ForEach(m => m.IdTelemetryUser = usersToNextId);
|
|
userFrom.IdUser = usersToNextId;
|
|
userFrom.IdTelemetry = to;
|
|
usersToNextId++;
|
|
}
|
|
}
|
|
|
|
var eventsFromQuery = db.TelemetryEvents
|
|
.Where(d => d.IdTelemetry == from);
|
|
var eventsFrom = await eventsFromQuery
|
|
.ToListAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
var eventsTo = await db.TelemetryEvents
|
|
.Where(d => d.IdTelemetry == to)
|
|
.AsNoTracking()
|
|
.ToListAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
var eventsToNextId = eventsTo.Max(e => e.IdEvent) + 1;
|
|
|
|
foreach (var eventFrom in eventsFrom)
|
|
{
|
|
var eventTo = eventsTo
|
|
.FirstOrDefault(e => e.IdEvent == eventFrom.IdEvent);
|
|
|
|
if (eventTo is null ||
|
|
eventTo.IdCategory != eventFrom.IdCategory ||
|
|
eventTo.MessageTemplate != eventFrom.MessageTemplate)
|
|
{
|
|
messagesFrom
|
|
.Where(m => m.IdEvent == eventFrom.IdEvent)
|
|
.ToList()
|
|
.ForEach(m => m.IdEvent = eventsToNextId);
|
|
eventFrom.IdEvent = eventsToNextId;
|
|
eventFrom.IdTelemetry = to;
|
|
eventsToNextId++;
|
|
}
|
|
}
|
|
|
|
await db.Database.ExecuteSqlRawAsync($"ALTER TABLE t_telemetry_user DISABLE TRIGGER ALL;", token).ConfigureAwait(false);
|
|
await db.Database.ExecuteSqlRawAsync($"ALTER TABLE t_telemetry_event DISABLE TRIGGER ALL;", token).ConfigureAwait(false);
|
|
await db.Database.ExecuteSqlRawAsync($"ALTER TABLE t_telemetry_message DISABLE TRIGGER ALL;", token).ConfigureAwait(false);
|
|
var affected = await db.SaveChangesAsync(token).ConfigureAwait(false);
|
|
await db.Database.ExecuteSqlRawAsync($"ALTER TABLE t_telemetry_user ENABLE TRIGGER ALL;", token).ConfigureAwait(false);
|
|
await db.Database.ExecuteSqlRawAsync($"ALTER TABLE t_telemetry_event ENABLE TRIGGER ALL;", token).ConfigureAwait(false);
|
|
await db.Database.ExecuteSqlRawAsync($"ALTER TABLE t_telemetry_message ENABLE TRIGGER ALL;", token).ConfigureAwait(false);
|
|
db.TelemetryUsers.RemoveRange(usersFromQuery);
|
|
db.TelemetryEvents.RemoveRange(eventsFromQuery);
|
|
affected += await db.SaveChangesAsync(token).ConfigureAwait(false);
|
|
return affected;
|
|
}
|
|
|
|
private async Task<int> MergeDataAsync<TEntity>(int from, int to, CancellationToken token)
|
|
where TEntity: class, AsbCloudDb.Model.ITelemetryData
|
|
{
|
|
const string IdTelemetryColumnName = "\"id_telemetry\"";
|
|
var dbSet = db.Set<TEntity>();
|
|
var tableName = dbSet.GetTableName();
|
|
var columns = dbSet.GetColumnsNames().ToList();
|
|
var index = columns.FindIndex(c => c.ToLower() == IdTelemetryColumnName.ToLower());
|
|
if (index < 0)
|
|
return -5;
|
|
columns[index] = $"{to} as {IdTelemetryColumnName}";
|
|
var columnsString = string.Join(',', columns);
|
|
var sql = $"INSERT INTO {tableName} " +
|
|
$"select {columnsString} " +
|
|
$"from {tableName} " +
|
|
$"where {IdTelemetryColumnName} = {from};" +
|
|
$"delete from {tableName} where {IdTelemetryColumnName} = {from};";
|
|
|
|
var affected = await db.Database.ExecuteSqlRawAsync(sql, token)
|
|
.ConfigureAwait(false);
|
|
|
|
return affected;
|
|
}
|
|
}
|
|
}
|