add achievement service; not tested

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

@@ -1,25 +1,18 @@
 import { Observable, OperatorFunction, map, pipe, withLatestFrom } from "rxjs";
 
-import { isEarnedEvent } from "./meta";
-import { AchievementEvent, AchievementValidator } from "./types";
-import { mapProgressByName } from "./util";
+import { AchievementId, AchievementValidator, MetricId } from "./types";
 
 // computes the list of live achievements; those whose trigger conditions
 // aren't met are excluded from the active set
 function active(
-  status$: Observable<AchievementEvent[]>,
+  metrics$: Observable<ReadonlyMap<MetricId, number>>,
+  earned$: Observable<ReadonlySet<AchievementId>>,
 ): OperatorFunction<AchievementValidator[], AchievementValidator[]> {
   return pipe(
     // TODO: accept a configuration observable that completes without
     //       emission when the user has opted out of achievements
-    withLatestFrom(status$),
-    map(([monitors, log]) => {
-      // partition the log into progress and earned achievements
-      const progressByName = mapProgressByName(log);
-      const earnedByName = new Set(
-        log.filter((e) => isEarnedEvent(e)).map((e) => e.achievement.name),
-      );
-
+    withLatestFrom(metrics$, earned$),
+    map(([monitors, metrics, earned]) => {
       // compute list of active achievements
       const active = monitors.filter((m) => {
         // 🧠 the filters could be lifted into a function argument & delivered
@@ -27,11 +20,11 @@ function active(
 
         if (m.active === "until-earned") {
           // monitor disabled if already achieved
-          return !earnedByName.has(m.achievement);
+          return !earned.has(m.achievement);
         }
 
         // monitor disabled if outside of threshold
-        const progress = (m.active.metric && progressByName.get(m.active.metric)) || 0;
+        const progress = (m.active.metric && metrics.get(m.active.metric)) || 0;
         if (progress > (m.active.high ?? Number.POSITIVE_INFINITY)) {
           return false;
         } else if (progress < (m.active.low ?? 0)) {

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

@@ -69,4 +69,4 @@ function achievements(
   );
 }
 
-export { achievements as validate };
+export { achievements, achievements as validate };

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

@@ -0,0 +1,65 @@
+import { Observable, ReplaySubject, Subject, debounceTime, filter, map, startWith } from "rxjs";
+
+import { active } from "./achievement-manager";
+import { achievements } from "./achievement-processor";
+import { latestEarnedSet, latestMetrics } from "./latest-metrics";
+import { isEarnedEvent, isProgressEvent } from "./meta";
+import {
+  AchievementEarnedEvent,
+  AchievementEvent,
+  AchievementId,
+  AchievementProgressEvent,
+  AchievementValidator,
+  MetricId,
+  UserActionEvent,
+} from "./types";
+
+const ACHIEVEMENT_INITIAL_DEBOUNCE_MS = 100;
+
+export class AchievementService {
+  constructor(
+    validators$: Observable<AchievementValidator[]>,
+    events$: Observable<UserActionEvent>,
+    bufferSize: number = 1000,
+  ) {
+    this.achievements = new Subject<AchievementEvent>();
+    this.achievementLog = new ReplaySubject<AchievementEvent>(bufferSize);
+    this.achievements.subscribe(this.achievementLog);
+
+    const active$ = validators$.pipe(active(this.metrics$(), this.earned$()));
+
+    events$.pipe(achievements(active$, this.metrics$())).subscribe(this.achievements);
+  }
+
+  private readonly achievements: Subject<AchievementEvent>;
+  private readonly achievementLog: ReplaySubject<AchievementEvent>;
+
+  earned$(): Observable<Set<AchievementId>> {
+    return this.achievementLog.pipe(
+      filter((e) => isEarnedEvent(e)),
+      map((e) => e as AchievementEarnedEvent),
+      latestEarnedSet(),
+      startWith(new Set<AchievementId>()),
+      debounceTime(ACHIEVEMENT_INITIAL_DEBOUNCE_MS),
+    );
+  }
+
+  metrics$(): Observable<Map<MetricId, number>> {
+    return this.achievementLog.pipe(
+      filter((e) => isProgressEvent(e)),
+      map((e) => e as AchievementProgressEvent),
+      latestMetrics(),
+      startWith(new Map<MetricId, number>()),
+    );
+  }
+
+  /** emit all achievement events */
+  all$(): Observable<AchievementEvent> {
+    return this.achievementLog.asObservable();
+  }
+
+  /** emit achievement events received after subscription */
+  new$(): Observable<AchievementEvent> {
+    return this.achievements.asObservable();
+  }
+}

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

@@ -1,6 +1,6 @@
 import { OperatorFunction, map, filter, pipe, scan } from "rxjs";
 
-import { MetricId, AchievementProgressEvent } from "./types";
+import { MetricId, AchievementProgressEvent, AchievementId, AchievementEarnedEvent } from "./types";
 
 function latestProgressEvents(): OperatorFunction<
   AchievementProgressEvent,
@@ -54,4 +54,13 @@ function latestMetrics(): OperatorFunction<AchievementProgressEvent, Map<MetricI
   );
 }
 
-export { latestMetrics, latestProgressEvents };
+function latestEarnedSet(): OperatorFunction<AchievementEarnedEvent, Set<AchievementId>> {
+  return pipe(
+    scan((earned, captured) => {
+      earned.add(captured.achievement.name);
+      return earned;
+    }, new Set<AchievementId>()),
+  );
+}
+
+export { latestMetrics, latestProgressEvents, latestEarnedSet };