partial tests for achievement hub

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

@@ -1,9 +1,92 @@
+import { BehaviorSubject, ReplaySubject, Subject, firstValueFrom } from "rxjs";
+
+import { ConsoleLogService } from "../../platform/services/console-log.service";
+import { consoleSemanticLoggerProvider } from "../log";
+
+import { AchievementHub } from "./achievement-hub";
+import { ItemCreatedEarnedEvent } from "./examples/achievement-events";
+import {
+  TotallyAttachedAchievement,
+  TotallyAttachedValidator,
+} from "./examples/example-validators";
+import { itemAdded$ } from "./examples/user-events";
+import {
+  AchievementEarnedEvent,
+  AchievementEvent,
+  AchievementId,
+  AchievementProgressEvent,
+  AchievementValidator,
+  MetricId,
+  UserActionEvent,
+} from "./types";
+
+const testLog = consoleSemanticLoggerProvider(new ConsoleLogService(true), {});
+
 describe("AchievementHub", () => {
-  describe("earned$", () => {});
+  describe("all$", () => {
+    it("emits achievements constructor emissions", async () => {
+      const validators$ = new Subject<AchievementValidator[]>();
+      const events$ = new Subject<UserActionEvent>();
+      const achievements$ = new Subject<AchievementEvent>();
+      const hub = new AchievementHub(validators$, events$, achievements$);
+      const results$ = new ReplaySubject<AchievementEvent>(3);
+      hub.all$().subscribe(results$);
 
-  describe("metrics$", () => {});
+      achievements$.next(ItemCreatedEarnedEvent);
 
-  describe("all$", () => {});
+      const result = firstValueFrom(results$);
+      await expect(result).resolves.toEqual(ItemCreatedEarnedEvent);
+    });
 
-  describe("named$", () => {});
+    it("emits achievements derived from events", async () => {
+      const validators$ = new BehaviorSubject<AchievementValidator[]>([TotallyAttachedValidator]);
+      const events$ = new Subject<UserActionEvent>();
+      const achievements$ = new Subject<AchievementEvent>();
+      const hub = new AchievementHub(validators$, events$, achievements$, 10, testLog);
+      const results$ = new ReplaySubject<AchievementEvent>(3);
+      hub.all$().subscribe(results$);
+
+      // hub starts listening when achievements$ completes
+      achievements$.complete();
+      itemAdded$.subscribe(events$);
+
+      const result = firstValueFrom(results$);
+      await expect(result).resolves.toMatchObject({
+        achievement: { type: "earned", name: TotallyAttachedAchievement },
+      });
+    });
+  });
+
+  describe("new$", () => {
+    it("", async () => {
+      const validators$ = new Subject<AchievementValidator[]>();
+      const events$ = new Subject<UserActionEvent>();
+      const achievements$ = new Subject<AchievementEvent>();
+      const hub = new AchievementHub(validators$, events$, achievements$);
+      const results$ = new ReplaySubject<AchievementEvent>(3);
+      hub.new$().subscribe(results$);
+    });
+  });
+
+  describe("earned$", () => {
+    it("", async () => {
+      const validators$ = new Subject<AchievementValidator[]>();
+      const events$ = new Subject<UserActionEvent>();
+      const achievements$ = new Subject<AchievementEvent>();
+      const hub = new AchievementHub(validators$, events$, achievements$);
+      const results$ = new ReplaySubject<Map<AchievementId, AchievementEarnedEvent>>(1);
+      hub.earned$().subscribe(results$);
+    });
+  });
+
+  describe("metrics$", () => {
+    it("", async () => {
+      const validators$ = new Subject<AchievementValidator[]>();
+      const events$ = new Subject<UserActionEvent>();
+      const achievements$ = new Subject<AchievementEvent>();
+      const hub = new AchievementHub(validators$, events$, achievements$);
+      const results$ = new ReplaySubject<Map<MetricId, AchievementProgressEvent>>(1);
+      hub.metrics$().subscribe(results$);
+    });
+  });
 });

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

@@ -2,13 +2,17 @@ import {
   Observable,
   ReplaySubject,
   Subject,
+  concat,
   debounceTime,
   filter,
   map,
-  share,
+  shareReplay,
   startWith,
+  tap,
 } from "rxjs";
 
+import { SemanticLogger, disabledSemanticLoggerProvider } from "../log";
+
 import { active } from "./achievement-manager";
 import { achievements } from "./achievement-processor";
 import { latestEarnedMetrics, latestProgressMetrics } from "./latest-metrics";
@@ -26,10 +30,26 @@ import {
 const ACHIEVEMENT_INITIAL_DEBOUNCE_MS = 100;
 
 export class AchievementHub {
+  /** Instantiates the achievement hub. A new achievement hub should be created
+   *   per-user, and streams should be partitioned by user.
+   *  @param validators$ emits the most recent achievement validator list and
+   *     re-emits the full list when the validators change.
+   *  @param events$ emits events captured from the system as they occur. THIS
+   *     OBSERVABLE IS SUBSCRIBED DURING INITIALIZATION. It must emit a complete
+   *     event to prevent the event hub from leaking the subscription.
+   *  @param achievements$ emits the list of achievement events captured before
+   *     initialization and then completes.  THIS OBSERVABLE IS SUBSCRIBED DURING
+   *     INITIALIZATION. Achievement processing begins once this observable
+   *     completes.
+   *  @param bufferSize the maximum number of achievement events retained by the
+   *     achievement hub.
+   */
   constructor(
     validators$: Observable<AchievementValidator[]>,
     events$: Observable<UserActionEvent>,
+    achievements$: Observable<AchievementEvent>,
     bufferSize: number = 1000,
+    private log: SemanticLogger = disabledSemanticLoggerProvider({}),
   ) {
     this.achievements = new Subject<AchievementEvent>();
     this.achievementLog = new ReplaySubject<AchievementEvent>(bufferSize);
@@ -37,36 +57,22 @@ export class AchievementHub {
 
     const metrics$ = this.metrics$().pipe(
       map((m) => new Map(Array.from(m.entries(), ([k, v]) => [k, v.achievement.value] as const))),
-      share(),
+      shareReplay({ bufferSize: 1, refCount: true }),
     );
     const earned$ = this.earned$().pipe(map((m) => new Set(m.keys())));
     const active$ = validators$.pipe(active(metrics$, earned$));
 
-    events$.pipe(achievements(active$, metrics$)).subscribe(this.achievements);
+    // TODO: figure out how to to unsubscribe from the event stream;
+    //   this likely requires accepting an account-bound observable, which
+    //   would also let the hub maintain it's "one user" invariant.
+    concat(achievements$, events$.pipe(achievements(active$, metrics$))).subscribe(
+      this.achievements,
+    );
   }
 
   private readonly achievements: Subject<AchievementEvent>;
   private readonly achievementLog: ReplaySubject<AchievementEvent>;
 
-  earned$(): Observable<Map<AchievementId, AchievementEarnedEvent>> {
-    return this.achievementLog.pipe(
-      filter((e) => isEarnedEvent(e)),
-      map((e) => e as AchievementEarnedEvent),
-      latestEarnedMetrics(),
-      startWith(new Map<AchievementId, AchievementEarnedEvent>()),
-      debounceTime(ACHIEVEMENT_INITIAL_DEBOUNCE_MS),
-    );
-  }
-
-  metrics$(): Observable<Map<MetricId, AchievementProgressEvent>> {
-    return this.achievementLog.pipe(
-      filter((e) => isProgressEvent(e)),
-      map((e) => e as AchievementProgressEvent),
-      latestProgressMetrics(),
-      startWith(new Map<MetricId, AchievementProgressEvent>()),
-    );
-  }
-
   /** emit all achievement events */
   all$(): Observable<AchievementEvent> {
     return this.achievementLog.asObservable();
@@ -76,4 +82,26 @@ export class AchievementHub {
   new$(): Observable<AchievementEvent> {
     return this.achievements.asObservable();
   }
+
+  earned$(): Observable<Map<AchievementId, AchievementEarnedEvent>> {
+    return this.achievementLog.pipe(
+      filter((e) => isEarnedEvent(e)),
+      map((e) => e as AchievementEarnedEvent),
+      latestEarnedMetrics(),
+      debounceTime(ACHIEVEMENT_INITIAL_DEBOUNCE_MS),
+      tap((m) => this.log.debug(m, "earned achievements update")),
+      startWith(new Map<AchievementId, AchievementEarnedEvent>()),
+    );
+  }
+
+  metrics$(): Observable<Map<MetricId, AchievementProgressEvent>> {
+    return this.achievementLog.pipe(
+      filter((e) => isProgressEvent(e)),
+      map((e) => e as AchievementProgressEvent),
+      latestProgressMetrics(),
+      debounceTime(ACHIEVEMENT_INITIAL_DEBOUNCE_MS),
+      tap((m) => this.log.debug(m, "achievement metrics update")),
+      startWith(new Map<MetricId, AchievementProgressEvent>()),
+    );
+  }
 }

libs/common/src/tools/log/logger.ts

@@ -1,12 +1,13 @@
-import { BehaviorSubject, SubjectLike, filter, first, from, map, zip } from "rxjs";
+import { BehaviorSubject, SubjectLike, from, map, zip } from "rxjs";
 import { Primitive } from "type-fest";
 
-import { Account, AccountService } from "../../auth/abstractions/account.service";
+import { Account } from "../../auth/abstractions/account.service";
 import { AppIdService } from "../../platform/abstractions/app-id.service";
 import { PlatformUtilsService } from "../../platform/abstractions/platform-utils.service";
 import { UserActionEvent } from "../achievements/types";
 
 import { ServiceFormat, UserFormat, EcsEventType } from "./ecs-format";
+import { disabledSemanticLoggerProvider } from "./factory";
 import { SemanticLogger } from "./semantic-logger.abstraction";
 
 export abstract class UserEventLogProvider {
@@ -25,22 +26,15 @@ export class UserEventLogger {
   constructor(
     idService: AppIdService,
     utilService: PlatformUtilsService,
-    accountService: AccountService,
+    account: Account,
     private now: () => number,
-    private log: SemanticLogger,
     private events$: SubjectLike<UserActionEvent>,
+    private log: SemanticLogger = disabledSemanticLoggerProvider({}),
   ) {
-    zip(
-      from(idService.getAppId()),
-      from(utilService.getApplicationVersion()),
-      accountService.activeAccount$.pipe(
-        filter((account) => !!account),
-        first(),
-      ),
-    )
+    zip(from(idService.getAppId()), from(utilService.getApplicationVersion()))
       .pipe(
         map(
-          ([appId, version, account]) =>
+          ([appId, version]) =>
             ({
               event: {
                 kind: "event",