introduce latest metrics rx operators

libs/common/src/tools/achievements/achievement-processor.spec.ts

@@ -1,7 +1,6 @@
 import { BehaviorSubject, ReplaySubject, bufferCount, concat, first, firstValueFrom } from "rxjs";
 
 import { validate } from "./achievement-processor";
-import { ItemCreatedProgressEvent } from "./examples/achievement-events";
 import {
   ItemCreatedAchievement,
   ItemCreatedProgress,
@@ -11,13 +10,13 @@ import {
   TotallyAttachedValidator,
 } from "./examples/example-validators";
 import { itemAdded$, itemUpdated$ } from "./examples/user-events";
-import { AchievementEvent } from "./types";
+import { AchievementEvent, MetricId } from "./types";
 
 describe("event-processor", () => {
   describe("validate", () => {
     it("earns an achievement", async () => {
       const validators$ = new BehaviorSubject([TotallyAttachedValidator]);
-      const captured$ = new BehaviorSubject<AchievementEvent[]>([]);
+      const captured$ = new BehaviorSubject(new Map<MetricId, number>());
       const achievements$ = new ReplaySubject<AchievementEvent>(2);
       const result = firstValueFrom(achievements$.pipe(bufferCount(2)));
 
@@ -29,7 +28,7 @@ describe("event-processor", () => {
 
     it("tracks achievement progress", async () => {
       const validators$ = new BehaviorSubject([ItemCreatedTracker]);
-      const captured$ = new BehaviorSubject<AchievementEvent[]>([]);
+      const captured$ = new BehaviorSubject(new Map<MetricId, number>());
       const achievements$ = new ReplaySubject<AchievementEvent>(2);
       const result = firstValueFrom(achievements$.pipe(bufferCount(2)));
 
@@ -41,7 +40,7 @@ describe("event-processor", () => {
 
     it("updates achievement progress", async () => {
       const validators$ = new BehaviorSubject([ItemCreatedTracker]);
-      const captured$ = new BehaviorSubject<AchievementEvent[]>([ItemCreatedProgressEvent]);
+      const captured$ = new BehaviorSubject(new Map([[ItemCreatedProgress, 1]]));
       const achievements$ = new ReplaySubject<AchievementEvent>(2);
       const result = firstValueFrom(achievements$.pipe(bufferCount(2)));
 
@@ -53,7 +52,7 @@ describe("event-processor", () => {
 
     it("tracks achievement progress and earns an achievement", async () => {
       const validators$ = new BehaviorSubject([ItemCreatedValidator]);
-      const captured$ = new BehaviorSubject<AchievementEvent[]>([]);
+      const captured$ = new BehaviorSubject(new Map<MetricId, number>());
       const achievements$ = new ReplaySubject<AchievementEvent>(3);
       const result = firstValueFrom(achievements$.pipe(bufferCount(3)));
 
@@ -70,7 +69,7 @@ describe("event-processor", () => {
 
     it("skips records that fail the validator's filter criteria", async () => {
       const validators$ = new BehaviorSubject([ItemCreatedTracker]);
-      const captured$ = new BehaviorSubject<AchievementEvent[]>([]);
+      const captured$ = new BehaviorSubject(new Map<MetricId, number>());
       const achievements$ = new ReplaySubject<AchievementEvent>(2);
       const result = firstValueFrom(achievements$.pipe(bufferCount(2)));
 
@@ -85,7 +84,7 @@ describe("event-processor", () => {
 
     it("only emits when its validators return events", async () => {
       const validators$ = new BehaviorSubject([ItemCreatedTracker]);
-      const captured$ = new BehaviorSubject<AchievementEvent[]>([]);
+      const captured$ = new BehaviorSubject(new Map<MetricId, number>());
       const achievements$ = new BehaviorSubject<AchievementEvent | null | undefined>(undefined);
 
       // `ItemCreatedTracker` filters `itemUpdated$` emissions. There are no others

libs/common/src/tools/achievements/achievement-processor.ts

@@ -8,15 +8,14 @@ import {
   MetricId,
   UserActionEvent,
 } from "./types";
-import { mapProgressByName as toMetricMap } from "./util";
 
 /** Monitors a user activity stream to recognize achievements
  *  @param validators$ validators track achievement progress and award achievements
- *  @param captured$ the set of previously emitted achievement events
+ *  @param metrics$ the set of previously emitted achievement events
  */
 function achievements(
   validators$: Observable<AchievementValidator[]>,
-  captured$: Observable<AchievementEvent[]>,
+  metrics$: Observable<Map<MetricId, number>>,
 ): OperatorFunction<UserActionEvent, AchievementEvent> {
   return pipe(
     withLatestFrom(validators$),
@@ -25,11 +24,10 @@ function achievements(
       const triggered = monitors.filter((m) => m.trigger(action));
       return [action, triggered] as const;
     }),
-    withLatestFrom(captured$),
+    withLatestFrom(metrics$),
     // monitor achievements
-    concatMap(([[action, validators], captured]) => {
+    concatMap(([[action, validators], metrics]) => {
       const achievements: AchievementEvent[] = [];
-      const metrics = toMetricMap(captured);
       const progress = new Map<AchievementId, AchievementProgressEvent[]>();
 
       // collect measurements

libs/common/src/tools/achievements/examples/achievement-events.ts

@@ -1,7 +1,7 @@
 import { UserId } from "../../../types/guid";
 import { AchievementProgressEvent } from "../types";
 
-import { ItemCreatedProgress } from "./example-validators";
+import { CredentialGeneratedProgress, ItemCreatedProgress } from "./example-validators";
 
 const ItemCreatedProgressEvent: AchievementProgressEvent = {
   "@timestamp": Date.now(),
@@ -24,4 +24,50 @@ const ItemCreatedProgressEvent: AchievementProgressEvent = {
   },
 };
 
-export { ItemCreatedProgressEvent };
+const NextItemCreatedProgressEvent: AchievementProgressEvent = {
+  "@timestamp": Date.now() + 100,
+  event: {
+    kind: "metric",
+    category: "session",
+  },
+  achievement: { type: "progress", name: ItemCreatedProgress, value: 2 },
+  service: {
+    name: "extension",
+    type: "client",
+    node: {
+      name: "an-installation-identifier-for-this-client-instance",
+    },
+    environment: "local",
+    version: "2025.3.1-innovation-sprint",
+  },
+  user: {
+    id: "some-guid" as UserId,
+  },
+};
+
+const CredentialGeneratedProgressEvent: AchievementProgressEvent = {
+  "@timestamp": Date.now(),
+  event: {
+    kind: "metric",
+    category: "session",
+  },
+  achievement: { type: "progress", name: CredentialGeneratedProgress, value: 1 },
+  service: {
+    name: "extension",
+    type: "client",
+    node: {
+      name: "an-installation-identifier-for-this-client-instance",
+    },
+    environment: "local",
+    version: "2025.3.1-innovation-sprint",
+  },
+  user: {
+    id: "some-guid" as UserId,
+  },
+};
+
+export {
+  ItemCreatedProgressEvent,
+  NextItemCreatedProgressEvent as ItemCreatedProgress2Event,
+  CredentialGeneratedProgressEvent,
+};

libs/common/src/tools/achievements/examples/example-validators.ts

@@ -3,6 +3,7 @@ import { Type } from "../data";
 import { AchievementId, MetricId, AchievementValidator } from "../types";
 
 const ItemCreatedProgress = "item-quantity" as MetricId;
+const CredentialGeneratedProgress = "credential-generated" as MetricId;
 
 const TotallyAttachedAchievement = "totally-attached" as AchievementId;
 const ItemCreatedMetric = "item-created-metric" as AchievementId;
@@ -118,4 +119,5 @@ export {
   ThreeItemsCreatedValidator,
   FiveItemsCreatedAchievement,
   FiveItemsCreatedValidator,
+  CredentialGeneratedProgress,
 };

libs/common/src/tools/achievements/latest-metrics.spec.ts

@@ -0,0 +1,66 @@
+import { BehaviorSubject, Subject } from "rxjs";
+
+import {
+  CredentialGeneratedProgressEvent,
+  ItemCreatedProgressEvent,
+  ItemCreatedProgress2Event,
+} from "./examples/achievement-events";
+import { CredentialGeneratedProgress, ItemCreatedProgress } from "./examples/example-validators";
+import { latestMetrics } from "./latest-metrics";
+import { AchievementProgressEvent, MetricId } from "./types";
+
+describe("latestMetrics", () => {
+  it("creates a map containing a metric", () => {
+    const subject = new Subject<AchievementProgressEvent>();
+    const result = new BehaviorSubject(new Map<MetricId, number>());
+
+    subject.pipe(latestMetrics()).subscribe(result);
+    subject.next(ItemCreatedProgressEvent);
+
+    expect(result.value.get(ItemCreatedProgress)).toEqual(
+      ItemCreatedProgressEvent.achievement.value,
+    );
+  });
+
+  it("creates a map containing multiple metrics", () => {
+    const subject = new Subject<AchievementProgressEvent>();
+    const result = new BehaviorSubject(new Map<MetricId, number>());
+
+    subject.pipe(latestMetrics()).subscribe(result);
+    subject.next(ItemCreatedProgressEvent);
+    subject.next(CredentialGeneratedProgressEvent);
+
+    expect(result.value.get(ItemCreatedProgress)).toEqual(
+      ItemCreatedProgressEvent.achievement.value,
+    );
+    expect(result.value.get(CredentialGeneratedProgress)).toEqual(
+      CredentialGeneratedProgressEvent.achievement.value,
+    );
+  });
+
+  it("creates a map containing updated metrics", () => {
+    const subject = new Subject<AchievementProgressEvent>();
+    const result = new BehaviorSubject(new Map<MetricId, number>());
+
+    subject.pipe(latestMetrics()).subscribe(result);
+    subject.next(ItemCreatedProgressEvent);
+    subject.next(ItemCreatedProgress2Event);
+
+    expect(result.value.get(ItemCreatedProgress)).toEqual(
+      ItemCreatedProgress2Event.achievement.value,
+    );
+  });
+
+  it("omits old events", () => {
+    const subject = new Subject<AchievementProgressEvent>();
+    const result = new BehaviorSubject(new Map<MetricId, number>());
+
+    subject.pipe(latestMetrics()).subscribe(result);
+    subject.next(ItemCreatedProgress2Event);
+    subject.next(ItemCreatedProgressEvent);
+
+    expect(result.value.get(ItemCreatedProgress)).toEqual(
+      ItemCreatedProgress2Event.achievement.value,
+    );
+  });
+});

libs/common/src/tools/achievements/latest-metrics.ts

@@ -0,0 +1,57 @@
+import { OperatorFunction, map, filter, pipe, scan } from "rxjs";
+
+import { MetricId, AchievementProgressEvent } from "./types";
+
+function latestProgressEvents(): OperatorFunction<
+  AchievementProgressEvent,
+  AchievementProgressEvent
+> {
+  type Accumulator = {
+    latest: Map<MetricId, AchievementProgressEvent>;
+    captured?: AchievementProgressEvent;
+  };
+  const acc: Accumulator = { latest: new Map() };
+
+  return pipe(
+    scan((acc, captured) => {
+      const { latest } = acc;
+      const current = latest.get(captured.achievement.name);
+
+      // omit stale events
+      if (current && current["@timestamp"] > captured["@timestamp"]) {
+        return { latest };
+      }
+
+      latest.set(captured.achievement.name, captured);
+      return { latest, captured };
+    }, acc),
+    // omit updates caused by stale events
+    filter(({ captured }) => !!captured),
+    map(({ captured }) => captured!),
+  );
+}
+
+function latestMetrics(): OperatorFunction<AchievementProgressEvent, Map<MetricId, number>> {
+  return pipe(
+    scan((metrics, captured) => {
+      const [timestamp] = metrics.get(captured.achievement.name) ?? [];
+
+      // omit stale metrics
+      if (timestamp && timestamp > captured["@timestamp"]) {
+        return metrics;
+      }
+
+      const latest = [captured["@timestamp"], captured.achievement.value] as const;
+      metrics.set(captured.achievement.name, latest);
+
+      return metrics;
+    }, new Map<MetricId, readonly [number, number]>()),
+
+    // omit timestamps from metrics
+    map(
+      (metrics) => new Map(Array.from(metrics.entries(), ([metric, [, value]]) => [metric, value])),
+    ),
+  );
+}
+
+export { latestMetrics, latestProgressEvents };