feature(entitlements): CR fixes

src/clients/entitlements/entitlements-client.ts

@@ -126,7 +126,7 @@ export class EntitlementsClient extends TypedEmitter<IEntitlementsClientEvents>
     const entitlementsData = await this.httpClient.get<VendorEntitlementsDto>('/api/v1/vendor-entitlements');
     const vendorEntitlementsDto = entitlementsData.data;
 
-    const { isUpdated, revision } = await this.cacheManager.loadSnapshotAsCurrent(vendorEntitlementsDto);
+    const { isUpdated, revision } = await this.cacheManager.loadSnapshotAsCurrentRevision(vendorEntitlementsDto);
 
     if (isUpdated) {
       this.emit(EntitlementsClientEventsEnum.SNAPSHOT_UPDATED, revision);
@@ -186,4 +186,4 @@ export class EntitlementsClient extends TypedEmitter<IEntitlementsClientEvents>
   destroy(): void {
     this.refreshTimeout && clearTimeout(this.refreshTimeout);
   }
-}
+}

src/clients/entitlements/storage/cache.revision-manager.spec.ts

@@ -68,7 +68,7 @@ describe(CacheRevisionManager.name, () => {
           jest.mocked(FronteggEntitlementsCacheInitializer.forLeader).mockResolvedValue(expectedNewEntitlementsCache);
 
           // when
-          loadingSnapshotResult = await cut.loadSnapshotAsCurrent(getDTO(333));
+          loadingSnapshotResult = await cut.loadSnapshotAsCurrentRevision(getDTO(333));
         });
 
         it('then it resolves to IsUpdatedToRev structure telling with updated revision.', async () => {
@@ -97,7 +97,7 @@ describe(CacheRevisionManager.name, () => {
           jest.mocked(FronteggEntitlementsCacheInitializer.forFollower).mockClear();
 
           // when
-          loadingSnapshotResult = await cut.loadSnapshotAsCurrent(getDTO(1));
+          loadingSnapshotResult = await cut.loadSnapshotAsCurrentRevision(getDTO(1));
         });
 
         it('then it resolves to IsUpdatedToRev structure telling nothing got updated and revision (1).', async () => {

src/clients/entitlements/storage/cache.revision-manager.ts

@@ -15,7 +15,7 @@ export class CacheRevisionManager {
 
   constructor(private readonly cache: ICacheManager<CacheValue>) {}
 
-  async loadSnapshotAsCurrent(dto: VendorEntitlementsDto): Promise<IsUpdatedToRev> {
+  async loadSnapshotAsCurrentRevision(dto: VendorEntitlementsDto): Promise<IsUpdatedToRev> {
     const currentRevision = await this.getCurrentCacheRevision();
     const givenRevision = dto.snapshotOffset;
 

src/clients/entitlements/storage/dto-to-cache-sources.mapper.ts

@@ -1,8 +1,32 @@
 import { FeatureId, VendorEntitlementsDto } from '../types';
 import { BundlesSource, ExpirationTime, FeatureSource, NO_EXPIRE, UNBUNDLED_SRC_ID } from './types';
 
+function ensureMapInMap<K, T extends Map<any, any>>(map: Map<K, T>, mapKey: K): T {
+  if (!map.has(mapKey)) {
+    map.set(mapKey, new Map() as T);
+  }
+
+  return map.get(mapKey)!;
+}
+
+function ensureArrayInMap<K, T>(map: Map<K, T[]>, mapKey: K): T[] {
+  if (!map.has(mapKey)) {
+    map.set(mapKey, []);
+  }
+
+  return map.get(mapKey)!;
+}
+
+function parseExpirationTime(time?: string | null): ExpirationTime {
+  if (time !== undefined && time !== null) {
+    return new Date(time).getTime();
+  }
+
+  return NO_EXPIRE;
+}
+
 export class DtoToCacheSourcesMapper {
-  map(dto: VendorEntitlementsDto): BundlesSource {
+  static map(dto: VendorEntitlementsDto): BundlesSource {
     const {
       data: { features, entitlements, featureBundles },
     } = dto;
@@ -56,15 +80,15 @@ export class DtoToCacheSourcesMapper {
       if (bundle) {
         if (userId) {
           // that's user-targeted entitlement
-          const tenantUserEntitlements = this.ensureMapInMap(bundle.user_entitlements, tenantId);
-          const usersEntitlements = this.ensureArrayInMap(tenantUserEntitlements, userId);
+          const tenantUserEntitlements = ensureMapInMap(bundle.user_entitlements, tenantId);
+          const usersEntitlements = ensureArrayInMap(tenantUserEntitlements, userId);
 
-          usersEntitlements.push(this.parseExpirationTime(expirationDate));
+          usersEntitlements.push(parseExpirationTime(expirationDate));
         } else {
           // that's tenant-targeted entitlement
-          const tenantEntitlements = this.ensureArrayInMap(bundle.tenant_entitlements, tenantId);
+          const tenantEntitlements = ensureArrayInMap(bundle.tenant_entitlements, tenantId);
 
-          tenantEntitlements.push(this.parseExpirationTime(expirationDate));
+          tenantEntitlements.push(parseExpirationTime(expirationDate));
         }
       } else {
         // TODO: issue warning here!
@@ -87,28 +111,4 @@ export class DtoToCacheSourcesMapper {
 
     return bundlesMap;
   }
-
-  private ensureMapInMap<K, T extends Map<any, any>>(map: Map<K, T>, mapKey: K): T {
-    if (!map.has(mapKey)) {
-      map.set(mapKey, new Map() as T);
-    }
-
-    return map.get(mapKey)!;
-  }
-
-  private ensureArrayInMap<K, T>(map: Map<K, T[]>, mapKey: K): T[] {
-    if (!map.has(mapKey)) {
-      map.set(mapKey, []);
-    }
-
-    return map.get(mapKey)!;
-  }
-
-  private parseExpirationTime(time?: string | null): ExpirationTime {
-    if (time !== undefined && time !== null) {
-      return new Date(time).getTime();
-    }
-
-    return NO_EXPIRE;
-  }
 }

src/clients/entitlements/storage/frontegg-cache/frontegg.cache-initializer.ts

@@ -25,7 +25,7 @@ export class FronteggEntitlementsCacheInitializer {
 
     const cacheInitializer = new FronteggEntitlementsCacheInitializer(entitlementsCache);
 
-    const sources = new DtoToCacheSourcesMapper().map(dto);
+    const sources = DtoToCacheSourcesMapper.map(dto);
 
     await cacheInitializer.setupPermissionsReadModel(sources);
     await cacheInitializer.setupEntitlementsReadModel(sources);
@@ -96,7 +96,7 @@ export class FronteggEntitlementsCacheInitializer {
     const allPermissions = await cache.collection(PERMISSIONS_COLLECTION_LIST).getAll<string>();
 
     for (const permission of allPermissions) {
-      await cache.expire([ getPermissionMappingKey(permission)], FronteggEntitlementsCacheInitializer.CLEAR_TTL);
+      await cache.expire([getPermissionMappingKey(permission)], FronteggEntitlementsCacheInitializer.CLEAR_TTL);
     }
 
     // clear static fields

src/components/leader-election/ioredis.lock-handler.ts

@@ -9,19 +9,43 @@ export class IORedisLockHandler implements ILockHandler {
   private static EXTEND_LEADERSHIP_SCRIPT =
     "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end";
 
-  async tryToMaintainTheLock(key: string, value: string, expirationTimeMs: number): Promise<boolean> {
+  /**
+   * This method calls the Lua script that prolongs the lock on given `leadershipResourceKey` only, when stored value
+   * equals to given `instanceIdentifier` and then method resolves to `true`.
+   *
+   * When `leadershipResourceKey` doesn't exist, or it has a different value, then the leadership is not prolonged and
+   * method resolves to `false`.
+   *
+   * Using Lua script ensures the atomicity of the whole process. Without it there is no guarantee that other Redis
+   * client doesn't execute operation on `leadershipResourceKey` in-between `GET` and `PEXPIRE` commands.
+   */
+  async tryToMaintainTheLock(
+    leadershipResourceKey: string,
+    instanceIdentifier: string,
+    expirationTimeMs: number,
+  ): Promise<boolean> {
     const extended = await this.redis.eval(
       IORedisLockHandler.EXTEND_LEADERSHIP_SCRIPT,
       NUM_OF_KEYS_IN_LUA_SCRIPT,
-      key,
-      value,
+      leadershipResourceKey,
+      instanceIdentifier,
       expirationTimeMs,
     );
 
     return (extended as number) > 0;
   }
 
-  async tryToLockLeaderResource(key: string, value: string, expirationTimeMs: number): Promise<boolean> {
-    return (await this.redis.set(key, value, 'PX', expirationTimeMs, 'NX')) !== null;
+  /**
+   * This stores the `instanceIdentifier` value into `leadershipResourceKey` only, when the key doesn't exist. If value
+   * is stored, then TTL is also set to `expirationTimeMs` and method resolves to `true`.
+   *
+   * Otherwise method resolved to `false` and no change to `leadershipResourceKey` is introduced.
+   */
+  async tryToLockLeaderResource(
+    leadershipResourceKey: string,
+    instanceIdentifier: string,
+    expirationTimeMs: number,
+  ): Promise<boolean> {
+    return (await this.redis.set(leadershipResourceKey, instanceIdentifier, 'PX', expirationTimeMs, 'NX')) !== null;
   }
 }

src/components/leader-election/redis-based.lock-handlers.spec.ts

@@ -61,7 +61,10 @@ describe.each([
       await expect(testRedisConnection.get(RESOURCE_KEY)).resolves.toEqual('bar');
 
       // and: key is about to expire
-      await expect(testRedisConnection.pttl(RESOURCE_KEY)).resolves.toBeWithin(0, 1000);
+      const pttl = await testRedisConnection.pttl(RESOURCE_KEY);
+
+      expect(pttl).toBeGreaterThan(0);
+      expect(pttl).toBeLessThanOrEqual(1000);
     });
 
     it('when .tryToMaintainTheLock(..) is called, then it resolves to FALSE and no value is written to resource key.', async () => {

src/components/leader-election/redis.lock-handler.ts

@@ -7,16 +7,42 @@ export class RedisLockHandler implements ILockHandler {
   private static EXTEND_LEADERSHIP_SCRIPT =
     "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end";
 
-  async tryToMaintainTheLock(key: string, value: string, expirationTimeMs: number): Promise<boolean> {
+  /**
+   * This method calls the Lua script that prolongs the lock on given `leadershipResourceKey` only, when stored value
+   * equals to given `instanceIdentifier` and then method resolves to `true`.
+   *
+   * When `leadershipResourceKey` doesn't exist, or it has a different value, then the leadership is not prolonged and
+   * method resolves to `false`.
+   *
+   * Using Lua script ensures the atomicity of the whole process. Without it there is no guarantee that other Redis
+   * client doesn't execute operation on `leadershipResourceKey` in-between `GET` and `PEXPIRE` commands.
+   */
+  async tryToMaintainTheLock(
+    leadershipResourceKey: string,
+    instanceIdentifier: string,
+    expirationTimeMs: number,
+  ): Promise<boolean> {
     const extended = await this.redis.EVAL(RedisLockHandler.EXTEND_LEADERSHIP_SCRIPT, {
-      keys: [key],
-      arguments: [value, expirationTimeMs.toString()],
+      keys: [leadershipResourceKey],
+      arguments: [instanceIdentifier, expirationTimeMs.toString()],
     });
 
     return (extended as number) > 0;
   }
 
-  async tryToLockLeaderResource(key: string, value: string, expirationTimeMs: number): Promise<boolean> {
-    return (await this.redis.SET(key, value, { PX: expirationTimeMs, NX: true })) !== null;
+  /**
+   * This stores the `instanceIdentifier` value into `leadershipResourceKey` only, when the key doesn't exist. If value
+   * is stored, then TTL is also set to `expirationTimeMs` and method resolves to `true`.
+   *
+   * Otherwise method resolved to `false` and no change to `leadershipResourceKey` is introduced.
+   */
+  async tryToLockLeaderResource(
+    leadershipResourceKey: string,
+    instanceIdentifier: string,
+    expirationTimeMs: number,
+  ): Promise<boolean> {
+    return (
+      (await this.redis.SET(leadershipResourceKey, instanceIdentifier, { PX: expirationTimeMs, NX: true })) !== null
+    );
   }
 }

src/components/leader-election/types.ts

@@ -1,6 +1,27 @@
 export interface ILockHandler {
-  tryToLockLeaderResource(key: string, value: string, expirationTimeMs: number): Promise<boolean>;
-  tryToMaintainTheLock(key: string, value: string, expirationTimeMs: number): Promise<boolean>;
+  /**
+   * This method is about to lock the `leadershipResourceKey` by writing its `instanceIdentifier` to it. The lock should
+   * not be permanent, but limited to given `expirationTimeMs`. Then the lock can be kept (extended) by calling
+   * `tryToMaintainTheLock` method.
+   */
+  tryToLockLeaderResource(
+    leadershipResourceKey: string,
+    instanceIdentifier: string,
+    expirationTimeMs: number,
+  ): Promise<boolean>;
+
+  /**
+   * This method is about to prolong the `leadershipResourceKey` time-to-live only, when the key contains value equal to
+   * given `instanceIdentifier`. Each instance competing for a leadership role needs to have a unique identifier.
+   *
+   * This way we know, that only the leader process can prolong its leadership. If leader dies, for any reason, no other
+   * process can extend its leadership.
+   */
+  tryToMaintainTheLock(
+    leadershipResourceKey: string,
+    instanceIdentifier: string,
+    expirationTimeMs: number,
+  ): Promise<boolean>;
 }
 
 export interface ILeadershipElectionOptions {