83 lines
2.6 KiB
Go
83 lines
2.6 KiB
Go
|
package orchestration
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"strconv"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/redis/go-redis/v9"
|
||
|
|
)
|
||
|
|
|
||
|
|
type LeaseRecord struct {
|
||
|
|
ConversationID string
|
||
|
|
DeviceID string
|
||
|
|
SequenceNo int64
|
||
|
|
Status string
|
||
|
|
ExpiresAt time.Time
|
||
|
|
ObservedAt time.Time
|
||
|
|
}
|
||
|
|
|
||
|
|
type LeaseCache interface {
|
||
|
|
Refresh(context.Context, LeaseRecord) error
|
||
|
|
End(context.Context, string, string, time.Time) error
|
||
|
|
}
|
||
|
|
|
||
|
|
type RedisLeaseCache struct {
|
||
|
|
client *redis.Client
|
||
|
|
prefix string
|
||
|
|
}
|
||
|
|
|
||
|
|
var refreshLeaseScript = redis.NewScript(`
|
||
|
|
local current = redis.call('HGET', KEYS[1], 'sequence_no')
|
||
|
|
if current and tonumber(current) >= tonumber(ARGV[1]) then
|
||
|
|
return 0
|
||
|
|
end
|
||
|
|
redis.call('HSET', KEYS[1],
|
||
|
|
'sequence_no', ARGV[1], 'status', ARGV[2], 'device_id', ARGV[3],
|
||
|
|
'lease_expires_at', ARGV[4], 'observed_at', ARGV[5])
|
||
|
|
redis.call('PEXPIRE', KEYS[1], ARGV[6])
|
||
|
|
redis.call('HSET', KEYS[2],
|
||
|
|
'online', 'true', 'conversation_id', ARGV[7], 'sequence_no', ARGV[1],
|
||
|
|
'last_seen_at', ARGV[5])
|
||
|
|
redis.call('PEXPIRE', KEYS[2], ARGV[8])
|
||
|
|
return 1
|
||
|
|
`)
|
||
|
|
|
||
|
|
func NewRedisLeaseCache(client *redis.Client, prefix string) (*RedisLeaseCache, error) {
|
||
|
|
if client == nil || prefix == "" {
|
||
|
|
return nil, errors.New("redis client and lease key prefix are required")
|
||
|
|
}
|
||
|
|
return &RedisLeaseCache{client: client, prefix: prefix}, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (c *RedisLeaseCache) Refresh(ctx context.Context, record LeaseRecord) error {
|
||
|
|
ttl := time.Until(record.ExpiresAt)
|
||
|
|
if ttl <= 0 || record.SequenceNo < 1 || record.ConversationID == "" || record.DeviceID == "" {
|
||
|
|
return errors.New("active lease record is invalid or already expired")
|
||
|
|
}
|
||
|
|
presenceTTL := ttl * 2
|
||
|
|
_, err := refreshLeaseScript.Run(ctx, c.client, []string{
|
||
|
|
c.prefix + ":conversation:lease:" + record.ConversationID,
|
||
|
|
c.prefix + ":device:presence:" + record.DeviceID,
|
||
|
|
},
|
||
|
|
strconv.FormatInt(record.SequenceNo, 10), record.Status, record.DeviceID,
|
||
|
|
record.ExpiresAt.UTC().Format(time.RFC3339Nano), record.ObservedAt.UTC().Format(time.RFC3339Nano),
|
||
|
|
ttl.Milliseconds(), record.ConversationID, presenceTTL.Milliseconds(),
|
||
|
|
).Result()
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (c *RedisLeaseCache) End(ctx context.Context, conversationID, deviceID string, observedAt time.Time) error {
|
||
|
|
if conversationID == "" || deviceID == "" {
|
||
|
|
return errors.New("conversation and device identifiers are required")
|
||
|
|
}
|
||
|
|
leaseKey := c.prefix + ":conversation:lease:" + conversationID
|
||
|
|
presenceKey := c.prefix + ":device:presence:" + deviceID
|
||
|
|
pipe := c.client.TxPipeline()
|
||
|
|
pipe.Del(ctx, leaseKey)
|
||
|
|
pipe.HSet(ctx, presenceKey, "conversation_id", "", "status", "idle", "last_seen_at", observedAt.UTC().Format(time.RFC3339Nano))
|
||
|
|
_, err := pipe.Exec(ctx)
|
||
|
|
return err
|
||
|
|
}
|