fix(stats): evaluate achievements for updated KOReader sessions

KOReader page-stat uploads can grow an existing derived session as more page events arrive, but
only newly inserted sessions reached achievement evaluation. A session that crossed a threshold by
being updated showed progress past that threshold while the badge stayed locked.

Carry the updated sessions out of the repository instead of a bare count and emit for them too.
The oversized-upload branch emitted a backfill event that nothing listened for, so subscribe
achievement evaluation to it as well; those runs are serialized per user and a burst collapses into
one trailing re-run, since each backfill re-evaluates the whole catalogue. Awards from that branch
stay silent: an upload that large is a device catching up, and notifying per badge would burst a
toast for every historical unlock.

Fixes #991

server/src/modules/achievement/achievement.service.test.ts

@@ -1,6 +1,6 @@
 import { describe, it, expect, vi, beforeEach } from 'vitest';
 import { AchievementService } from './achievement.service';
-import { AchievementEventsService, ACHIEVEMENT_EVENT_BOOK_STATUS_CHANGED } from './achievement-events.service';
+import { AchievementEventsService, ACHIEVEMENT_EVENT_BACKFILL, ACHIEVEMENT_EVENT_BOOK_STATUS_CHANGED } from './achievement-events.service';
 
 function makeRepo() {
   return {
@@ -79,6 +79,44 @@ describe('AchievementService', () => {
       await service.onModuleInit();
       expect(repo.upsertCatalogue).toHaveBeenCalledOnce();
     });
+
+    it('subscribes achievement evaluation to backfill events', async () => {
+      const handleEvent = vi.spyOn(service, 'handleEvent').mockResolvedValue(undefined);
+      await service.onModuleInit();
+
+      events.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: 7 });
+
+      expect(handleEvent).toHaveBeenCalledWith(ACHIEVEMENT_EVENT_BACKFILL, { userId: 7 });
+    });
+
+    it('collapses a burst of backfill events for one user into a single trailing re-run', async () => {
+      const pending: (() => void)[] = [];
+      const handleEvent = vi.spyOn(service, 'handleEvent').mockImplementation(() => new Promise<void>((resolve) => pending.push(resolve)));
+      await service.onModuleInit();
+
+      events.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: 7 });
+      events.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: 7 });
+      events.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: 7 });
+      expect(handleEvent).toHaveBeenCalledOnce();
+
+      pending[0]!();
+      await vi.waitFor(() => expect(handleEvent).toHaveBeenCalledTimes(2));
+
+      pending[1]!();
+      await new Promise((resolve) => setTimeout(resolve, 0));
+      expect(handleEvent).toHaveBeenCalledTimes(2);
+    });
+
+    it('does not serialize backfill runs across users', async () => {
+      const handleEvent = vi.spyOn(service, 'handleEvent').mockImplementation(() => new Promise<void>(() => undefined));
+      await service.onModuleInit();
+
+      events.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: 7 });
+      events.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: 8 });
+
+      expect(handleEvent).toHaveBeenCalledTimes(2);
+      expect(handleEvent).toHaveBeenCalledWith(ACHIEVEMENT_EVENT_BACKFILL, { userId: 8 });
+    });
   });
 
   describe('getCatalogue', () => {
@@ -585,6 +623,16 @@ describe('AchievementService', () => {
       expect(registry.evaluate).not.toHaveBeenCalled();
     });
 
+    it('awards backfill achievements without notifying', async () => {
+      registry.evaluate.mockResolvedValueOnce([{ key: 'power_hour', context: { pages: 112 } }]).mockResolvedValueOnce([]);
+      repo.findAchievementByKey.mockResolvedValue({ key: 'power_hour', name: 'Power Hour', rarity: 'rare', iconName: 'zap' });
+
+      await service.handleEvent(ACHIEVEMENT_EVENT_BACKFILL, { userId: 1 });
+
+      expect(repo.award).toHaveBeenCalledWith(1, 'power_hour', { pages: 112 });
+      expect(notificationService.notify).not.toHaveBeenCalled();
+    });
+
     it('handles achievement preference lookup errors gracefully', async () => {
       userService.isAchievementEnabled.mockRejectedValue(new Error('Settings lookup failed'));
 
@@ -604,7 +652,7 @@ describe('AchievementService', () => {
       await service.onModuleInit();
       const removeListenerSpy = vi.spyOn(events, 'removeListener');
       service.onModuleDestroy();
-      expect(removeListenerSpy).toHaveBeenCalledTimes(7);
+      expect(removeListenerSpy).toHaveBeenCalledTimes(8);
     });
   });
 });

server/src/modules/achievement/achievement.service.ts

@@ -27,6 +27,7 @@ import type { AchievementRow, UserAchievementRow } from '../../db/schema';
 export class AchievementService implements OnModuleInit, OnModuleDestroy {
   private readonly logger = new Logger(AchievementService.name);
   private readonly eventHandlers = new Map<string, (payload: Record<string, unknown>) => void>();
+  private readonly backfillRuns = new Map<number, { pending: boolean }>();
   private seedPromise: Promise<void> | null = null;
 
   constructor(
@@ -195,6 +196,44 @@ export class AchievementService implements OnModuleInit, OnModuleDestroy {
       this.eventHandlers.set(eventName, handler);
       this.events.on(eventName, handler);
     }
+
+    const backfillHandler = (payload: Record<string, unknown>): void => {
+      this.handleBackfillEvent(payload);
+    };
+    this.eventHandlers.set(ACHIEVEMENT_EVENT_BACKFILL, backfillHandler);
+    this.events.on(ACHIEVEMENT_EVENT_BACKFILL, backfillHandler);
+  }
+
+  /**
+   * A backfill re-evaluates the whole catalogue and carries no data beyond the user, so emitters that
+   * fire it per unit of work (a KOReader device catching up over many uploads) would otherwise stack
+   * full evaluations for one user in parallel. Runs are serialized per user and a burst collapses into
+   * a single trailing re-run, which still sees the newest data because every evaluator reads the database.
+   */
+  private handleBackfillEvent(payload: Record<string, unknown>): void {
+    const userId = payload.userId as number;
+    if (!userId) return;
+
+    const active = this.backfillRuns.get(userId);
+    if (active) {
+      active.pending = true;
+      return;
+    }
+
+    const run = { pending: false };
+    this.backfillRuns.set(userId, run);
+    void this.drainBackfillRuns(userId, run, payload);
+  }
+
+  private async drainBackfillRuns(userId: number, run: { pending: boolean }, payload: Record<string, unknown>): Promise<void> {
+    try {
+      do {
+        run.pending = false;
+        await this.handleEvent(ACHIEVEMENT_EVENT_BACKFILL, payload);
+      } while (run.pending);
+    } finally {
+      this.backfillRuns.delete(userId);
+    }
   }
 
   private async sendNotification(userId: number, achievementKey: string): Promise<void> {

server/src/modules/koreader/koreader-plugin.repository.ts

@@ -36,7 +36,7 @@ export interface IngestPageStatsResult {
   accepted: number;
   duplicates: number;
   insertedSessions: DerivedKoreaderSession[];
-  updatedSessions: number;
+  updatedSessions: DerivedKoreaderSession[];
   deletedSessions: number;
 }
 
@@ -83,7 +83,7 @@ export class KoreaderPluginRepository {
       const accepted = inserted.length;
       const duplicates = events.length - accepted;
       if (accepted === 0) {
-        return { accepted, duplicates, insertedSessions: [], updatedSessions: 0, deletedSessions: 0 };
+        return { accepted, duplicates, insertedSessions: [], updatedSessions: [], deletedSessions: 0 };
       }
 
       // Only the clusters the new events belong to can change, so derivation reads and
@@ -225,7 +225,7 @@ export class KoreaderPluginRepository {
         accepted,
         duplicates,
         insertedSessions: toInsert,
-        updatedSessions: toUpdate.length,
+        updatedSessions: toUpdate,
         deletedSessions: toDelete.length,
       };
     });

server/src/modules/koreader/koreader-stats.service.test.ts

@@ -57,7 +57,9 @@ describe('KoreaderStatsService', () => {
       restoreDevice: vi.fn().mockResolvedValue(undefined),
     };
     pluginRepo = {
-      ingestAndDeriveForBook: vi.fn().mockResolvedValue({ accepted: 2, duplicates: 0, insertedSessions: [], updatedSessions: 0, deletedSessions: 0 }),
+      ingestAndDeriveForBook: vi
+        .fn()
+        .mockResolvedValue({ accepted: 2, duplicates: 0, insertedSessions: [], updatedSessions: [], deletedSessions: 0 }),
     };
     achievementEvents = { emit: vi.fn() };
 
@@ -101,7 +103,7 @@ describe('KoreaderStatsService', () => {
       accepted: 0,
       duplicates: 2,
       insertedSessions: [],
-      updatedSessions: 0,
+      updatedSessions: [],
       deletedSessions: 0,
     });
     const dto = makeDto([
@@ -124,7 +126,7 @@ describe('KoreaderStatsService', () => {
       accepted: 2,
       duplicates: 0,
       insertedSessions: [makeSession(1000), makeSession(5000)],
-      updatedSessions: 0,
+      updatedSessions: [],
       deletedSessions: 0,
     });
     const dto = makeDto([{ hash: HASH_A, events: [{ page: 1, startTime: 1000, durationSeconds: 30, totalPages: 100 }] }]);
@@ -138,12 +140,43 @@ describe('KoreaderStatsService', () => {
     );
   });
 
-  it('emits a single backfill event when more sessions than the threshold are inserted', async () => {
+  it('emits a reading-session event with the latest values for an updated session', async () => {
+    const updatedSession = {
+      ...makeSession(1000),
+      durationSeconds: 180,
+      progressDelta: 56,
+      endProgress: 56.5,
+    };
+    pluginRepo.ingestAndDeriveForBook.mockResolvedValue({
+      accepted: 1,
+      duplicates: 0,
+      insertedSessions: [],
+      updatedSessions: [updatedSession],
+      deletedSessions: 0,
+    });
+    const dto = makeDto([{ hash: HASH_A, events: [{ page: 113, startTime: 1200, durationSeconds: 60, totalPages: 200 }] }]);
+
+    await service.uploadPageStats(makeUser(), dto);
+
+    expect(achievementEvents.emit).toHaveBeenCalledOnce();
+    expect(achievementEvents.emit).toHaveBeenCalledWith(ACHIEVEMENT_EVENT_READING_SESSION_SAVED, {
+      userId: 7,
+      bookFileId: 10,
+      durationSeconds: 180,
+      startedAt: updatedSession.startedAt,
+      endedAt: updatedSession.endedAt,
+      progressDelta: 56,
+      endProgress: 56.5,
+      timezone: 'Asia/Kolkata',
+    });
+  });
+
+  it('emits a single backfill event when inserted and updated sessions together exceed the threshold', async () => {
     pluginRepo.ingestAndDeriveForBook.mockResolvedValue({
       accepted: 30,
       duplicates: 0,
-      insertedSessions: Array.from({ length: 21 }, (_, i) => makeSession(1000 + i * 4000)),
-      updatedSessions: 0,
+      insertedSessions: Array.from({ length: 11 }, (_, i) => makeSession(1000 + i * 4000)),
+      updatedSessions: Array.from({ length: 10 }, (_, i) => makeSession(50_000 + i * 4000)),
       deletedSessions: 0,
     });
     const dto = makeDto([{ hash: HASH_A, events: [{ page: 1, startTime: 1000, durationSeconds: 30, totalPages: 100 }] }]);

server/src/modules/koreader/koreader-stats.service.ts

@@ -58,8 +58,10 @@ export class KoreaderStatsService {
 
       const results: PageStatsBookResult[] = [];
       const unmatched: string[] = [];
-      const insertedSessions: { session: DerivedKoreaderSession; bookFileId: number }[] = [];
+      const changedSessions: { session: DerivedKoreaderSession; bookFileId: number }[] = [];
       let acceptedTotal = 0;
+      let insertedSessionCount = 0;
+      let updatedSessionCount = 0;
 
       for (const book of dto.books) {
         const hash = book.hash.toLowerCase();
@@ -80,8 +82,13 @@ export class KoreaderStatsService {
         });
 
         acceptedTotal += result.accepted;
+        insertedSessionCount += result.insertedSessions.length;
+        updatedSessionCount += result.updatedSessions.length;
         for (const session of result.insertedSessions) {
-          insertedSessions.push({ session, bookFileId: match.bookFileId });
+          changedSessions.push({ session, bookFileId: match.bookFileId });
+        }
+        for (const session of result.updatedSessions) {
+          changedSessions.push({ session, bookFileId: match.bookFileId });
         }
 
         // Watermark covers every event processed in this batch, duplicates included, so a plugin
@@ -92,10 +99,10 @@ export class KoreaderStatsService {
 
       if (acceptedTotal > 0) await this.koreaderRepo.restoreDevice(user.id, dto.deviceId);
 
-      this.emitSessionAchievements(user, insertedSessions);
+      this.emitSessionAchievements(user, changedSessions);
 
       this.logger.log(
-        `[${PAGE_STATS_EVENT}] [end] userId=${user.id} deviceId=${dto.deviceId.slice(0, 8)} durationMs=${Date.now() - startedAtMs} accepted=${acceptedTotal} sessionsInserted=${insertedSessions.length} unmatched=${unmatched.length} - page stats upload completed`,
+        `[${PAGE_STATS_EVENT}] [end] userId=${user.id} deviceId=${dto.deviceId.slice(0, 8)} durationMs=${Date.now() - startedAtMs} accepted=${acceptedTotal} sessionsInserted=${insertedSessionCount} sessionsUpdated=${updatedSessionCount} unmatched=${unmatched.length} - page stats upload completed`,
       );
 
       return { results, unmatched };
@@ -108,16 +115,19 @@ export class KoreaderStatsService {
     }
   }
 
-  private emitSessionAchievements(user: RequestUser, inserted: { session: DerivedKoreaderSession; bookFileId: number }[]) {
-    if (inserted.length === 0) return;
+  private emitSessionAchievements(user: RequestUser, sessions: { session: DerivedKoreaderSession; bookFileId: number }[]) {
+    if (sessions.length === 0) return;
 
-    if (inserted.length > KOREADER_BACKFILL_EVENT_THRESHOLD) {
+    // An upload this large is a device catching up rather than a live reading session, so it is
+    // evaluated as a backfill: one pass over the catalogue instead of one per session, and awards
+    // land silently. Notifying per badge here would burst a toast for every historical unlock.
+    if (sessions.length > KOREADER_BACKFILL_EVENT_THRESHOLD) {
       this.achievementEvents.emit(ACHIEVEMENT_EVENT_BACKFILL, { userId: user.id });
       return;
     }
 
     const timezone = resolveTimeZone((user.settings as unknown as UserSettings | undefined)?.timezone, 'UTC');
-    for (const { session, bookFileId } of inserted) {
+    for (const { session, bookFileId } of sessions) {
       this.achievementEvents.emit(ACHIEVEMENT_EVENT_READING_SESSION_SAVED, {
         userId: user.id,
         bookFileId,

server/test/koreader-page-stats.e2e-spec.ts

@@ -4,6 +4,7 @@ import { and, asc, eq, like } from 'drizzle-orm';
 import * as schema from '../src/db/schema';
 import {
   KOREADER_SESSION_GAP_SECONDS,
+  buildSessionId,
   buildSessionIdPrefix,
   deriveKoreaderSessions,
   type KoreaderPageEvent,
@@ -26,6 +27,8 @@ const KOREADER_PASSWORD = 'PageStatsDevicePass123';
 const DEVICE_ID = 'e2e-stats-0001';
 const BASE_EPOCH = 1_780_000_000;
 const TOTAL_PAGES = 300;
+/** Mirrors POWER_HOUR_PAGES in the reading evaluator. */
+const POWER_HOUR_PAGES = 75;
 
 /** A contiguous run of one-minute page reads starting at `startEpoch`. */
 function cluster(startEpoch: number, pages: number, firstPage: number): KoreaderPageEvent[] {
@@ -98,6 +101,28 @@ describe('KOReader page stats derivation (e2e)', { timeout: 180_000 }, () => {
       }));
   }
 
+  async function hasAchievement(userId: number, key: string): Promise<boolean> {
+    const rows = await ctx.db
+      .select({ key: schema.userAchievements.achievementKey })
+      .from(schema.userAchievements)
+      .where(and(eq(schema.userAchievements.userId, userId), eq(schema.userAchievements.achievementKey, key)));
+    return rows.length > 0;
+  }
+
+  /** Achievement evaluation is fire-and-forget off the upload response, so the award lands after it. */
+  async function waitForAchievement(userId: number, key: string, timeoutMs = 20_000): Promise<boolean> {
+    const deadline = Date.now() + timeoutMs;
+    for (;;) {
+      if (await hasAchievement(userId, key)) return true;
+      if (Date.now() >= deadline) return false;
+      await new Promise((resolve) => setTimeout(resolve, 100));
+    }
+  }
+
+  function pagesFor(progressDelta: number | null): number {
+    return ((progressDelta ?? 0) / 100) * TOTAL_PAGES;
+  }
+
   function comparable(rows: Awaited<ReturnType<typeof storedSessions>>) {
     return rows.map((row) => ({
       sessionId: row.sessionId,
@@ -203,4 +228,37 @@ describe('KOReader page stats derivation (e2e)', { timeout: 180_000 }, () => {
     expect(noop.results[0]!.accepted).toBe(0);
     expect(await storedSessions()).toEqual(complete);
   });
+
+  it('unlocks a session achievement when a later batch grows a stored session past its threshold', async () => {
+    const metadata = await ctx.db
+      .update(schema.bookMetadata)
+      .set({ pageCount: TOTAL_PAGES })
+      .where(eq(schema.bookMetadata.bookId, epub.bookId))
+      .returning({ bookId: schema.bookMetadata.bookId });
+    expect(metadata).toHaveLength(1);
+
+    const opening = cluster(BASE_EPOCH + 13_000_000, 40, 1);
+    await uploadPageStats(opening);
+
+    const sessionId = buildSessionId(DEVICE_ID, epub.bookFileId, opening[0]!.startTime);
+    const [session] = await ctx.db
+      .select({ userId: schema.readingSessions.userId, progressDelta: schema.readingSessions.progressDelta })
+      .from(schema.readingSessions)
+      .where(eq(schema.readingSessions.sessionId, sessionId));
+    expect(session).toBeTruthy();
+    expect(pagesFor(session!.progressDelta)).toBeLessThan(POWER_HOUR_PAGES);
+    expect(await hasAchievement(session!.userId, 'power_hour')).toBe(false);
+
+    // Resumes within the session gap, so the stored session grows under its existing id instead of
+    // being inserted again: the update path that previously emitted no achievement event.
+    const continued = cluster(opening.at(-1)!.startTime + 60, 45, opening.at(-1)!.page + 1);
+    await uploadPageStats(continued);
+
+    const grown = (await storedSessions()).filter((row) => row.startedAt.getTime() >= (BASE_EPOCH + 13_000_000) * 1000);
+    expect(grown).toHaveLength(1);
+    expect(grown[0]!.sessionId).toBe(sessionId);
+    expect(pagesFor(grown[0]!.progressDelta)).toBeGreaterThanOrEqual(POWER_HOUR_PAGES);
+
+    expect(await waitForAchievement(session!.userId, 'power_hour')).toBe(true);
+  });
 });