using IM.Commons; using MessageService.Infrastructure; using Microsoft.EntityFrameworkCore; using RedLockNet; using StackExchange.Redis; namespace MessageService.WebApi.Application { public class SquenceService { private readonly IDatabase _database; private readonly IDistributedLockFactory _lockFactory; private readonly MessageDbContext messageDb; public SquenceService(IConnectionMultiplexer multiplexer, IDistributedLockFactory distributedLockFactory, MessageDbContext messageDb ) { _database = multiplexer.GetDatabase(); _lockFactory = distributedLockFactory; this.messageDb = messageDb; } public async Task GetNextSquenceIdAsync(string streamKey) { string key = RedisHelper.GetSequenceIdKey(streamKey); string lockKey = RedisHelper.GetSequenceIdLockKey(streamKey); var exists = await _database.KeyExistsAsync(key); if (!exists) { using (var _lock = await _lockFactory.CreateLockAsync(lockKey, TimeSpan.FromSeconds(5))) { if (_lock.IsAcquired) { if (!await _database.KeyExistsAsync(key)) { var max = await messageDb.Messages .Where(x => x.StreamKey == streamKey) .MaxAsync(m => (long?)m.SequenceId) ?? 0; await _database.StringSetAsync(key, max, TimeSpan.FromDays(7)); } } } } return await _database.StringIncrementAsync(key); } } }