fix(kobo): derive reading sessions from the device reading-time counter

Current firmware never posts the LeaveContent analytics events the reading
log was fed from, so every Kobo statistic read zero. Derive sessions from the
SpentReadingMinutes counter the device already sends on each state push,
tracked per device in reading_session_sync_cursors so two readers of one book
cannot inflate each other. Analytics sessions still win and retire any
estimate they overlap.

Closes #1058
Closes #741

server/src/db/migrations/0078_add_kobo_statistics_cursors.sql

@@ -0,0 +1,19 @@
+CREATE TABLE "reading_session_sync_cursors" (
+	"user_id" integer NOT NULL,
+	"book_id" integer NOT NULL,
+	"source" varchar(32) NOT NULL,
+	"source_device_key" varchar(128) NOT NULL,
+	"counter" integer NOT NULL,
+	"generation" integer DEFAULT 0 NOT NULL,
+	"last_modified" timestamp with time zone NOT NULL,
+	"updated_at" timestamp with time zone DEFAULT now() NOT NULL,
+	CONSTRAINT "reading_session_sync_cursors_user_id_book_id_source_source_device_key_pk" PRIMARY KEY("user_id","book_id","source","source_device_key"),
+	CONSTRAINT "rssc_counter_nonnegative_chk" CHECK ("reading_session_sync_cursors"."counter" >= 0),
+	CONSTRAINT "rssc_generation_nonnegative_chk" CHECK ("reading_session_sync_cursors"."generation" >= 0)
+);
+--> statement-breakpoint
+ALTER TABLE "reading_sessions" ADD COLUMN "source_device_key" varchar(128);--> statement-breakpoint
+ALTER TABLE "reading_session_sync_cursors" ADD CONSTRAINT "reading_session_sync_cursors_user_id_users_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "reading_session_sync_cursors" ADD CONSTRAINT "reading_session_sync_cursors_book_id_books_id_fk" FOREIGN KEY ("book_id") REFERENCES "public"."books"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+CREATE INDEX "rssc_book_id_idx" ON "reading_session_sync_cursors" USING btree ("book_id");--> statement-breakpoint
+CREATE INDEX "rs_user_book_source_device_started_idx" ON "reading_sessions" USING btree ("user_id","book_id","source","source_device_key","started_at");

server/src/db/migrations/meta/_journal.json

@@ -547,6 +547,13 @@
       "when": 1787678883089,
       "tag": "0077_add_notification_coalescing",
       "breakpoints": true
+    },
+    {
+      "idx": 78,
+      "version": "7",
+      "when": 1787689487276,
+      "tag": "0078_add_kobo_statistics_cursors",
+      "breakpoints": true
     }
   ]
 }

server/src/db/schema/reader.test.ts

@@ -6,6 +6,7 @@ import {
   readerDefaultPreferences,
   readerPreferences,
   readingProgress,
+  readingSessionSyncCursors,
   readingSessions,
   userReadingDailyStats,
 } from './reader';
@@ -58,10 +59,21 @@ describe('reader schema', () => {
     expect(indexNames).toContain('rs_user_started_at_idx');
     expect(indexNames).toContain('rs_book_file_started_at_idx');
     expect(indexNames).toContain('rs_user_book_file_idx');
+    expect(indexNames).toContain('rs_user_book_source_device_started_idx');
     expect(fkMap.get('user_id')?.onDelete).toBe('cascade');
     expect(fkMap.get('book_file_id')?.onDelete).toBe('cascade');
   });
 
+  it('keeps cumulative sync cursors isolated by user, book, source, and device', () => {
+    const config = getTableConfig(readingSessionSyncCursors);
+    const pkColumns = config.primaryKeys.map((pk) => pk.columns.map((col) => col.name));
+    const fkMap = fkByColumn(readingSessionSyncCursors);
+
+    expect(pkColumns).toContainEqual(['user_id', 'book_id', 'source', 'source_device_key']);
+    expect(fkMap.get('user_id')?.onDelete).toBe('cascade');
+    expect(fkMap.get('book_id')?.onDelete).toBe('cascade');
+  });
+
   it('stores user daily reading aggregates keyed by user, library, and day', () => {
     const config = getTableConfig(userReadingDailyStats);
     const pkColumns = config.primaryKeys.map((pk) => pk.columns.map((col) => col.name));

server/src/db/schema/reader.ts

@@ -214,6 +214,7 @@ export const readingSessions = pgTable(
     sessionId: varchar('session_id', { length: 64 }).notNull(),
     // 'web' (browser reader) | 'koreader' (page-stats derivation) | 'manual' (user-entered) | 'kobo' (future)
     source: varchar('source', { length: 10 }).$type<ReadingSessionSource>(),
+    sourceDeviceKey: varchar('source_device_key', { length: 128 }),
     startedAt: timestamp('started_at', { withTimezone: true }).notNull(),
     endedAt: timestamp('ended_at', { withTimezone: true }).notNull(),
     // Server-computed from endedAt - startedAt; client-provided timestamps are untrusted for duration.
@@ -228,6 +229,7 @@ export const readingSessions = pgTable(
     index('rs_book_file_started_at_idx').on(t.bookFileId, t.startedAt),
     index('rs_user_book_file_idx').on(t.userId, t.bookFileId),
     index('rs_user_book_started_at_idx').on(t.userId, t.bookId, t.startedAt),
+    index('rs_user_book_source_device_started_idx').on(t.userId, t.bookId, t.source, t.sourceDeviceKey, t.startedAt),
     index('reading_sessions_book_id_idx').on(t.bookId),
     index('rs_attempt_started_at_idx').on(t.attemptId, t.startedAt),
     check('reading_sessions_source_chk', sql`${t.source} in ('web', 'koreader', 'manual', 'kobo')`),
@@ -240,6 +242,35 @@ export const readingSessions = pgTable(
 export type ReadingSession = typeof readingSessions.$inferSelect;
 export type NewReadingSession = typeof readingSessions.$inferInsert;
 
+export const readingSessionSyncCursors = pgTable(
+  'reading_session_sync_cursors',
+  {
+    userId: integer('user_id')
+      .notNull()
+      .references(() => users.id, { onDelete: 'cascade' }),
+    bookId: integer('book_id')
+      .notNull()
+      .references(() => books.id, { onDelete: 'cascade' }),
+    source: varchar('source', { length: 32 }).notNull(),
+    sourceDeviceKey: varchar('source_device_key', { length: 128 }).notNull(),
+    counter: integer('counter').notNull(),
+    generation: integer('generation').notNull().default(0),
+    lastModified: timestamp('last_modified', { withTimezone: true }).notNull(),
+    updatedAt: timestamp('updated_at', { withTimezone: true })
+      .defaultNow()
+      .notNull()
+      .$onUpdateFn(() => new Date()),
+  },
+  (t) => [
+    primaryKey({ columns: [t.userId, t.bookId, t.source, t.sourceDeviceKey] }),
+    index('rssc_book_id_idx').on(t.bookId),
+    check('rssc_counter_nonnegative_chk', sql`${t.counter} >= 0`),
+    check('rssc_generation_nonnegative_chk', sql`${t.generation} >= 0`),
+  ],
+);
+
+export type ReadingSessionSyncCursor = typeof readingSessionSyncCursors.$inferSelect;
+
 export const userReadingDailyStats = pgTable(
   'user_reading_daily_stats',
   {

server/src/modules/kobo/kobo-statistics-session.util.test.ts

@@ -0,0 +1,33 @@
+import {
+  KOBO_STATISTICS_SESSION_ID_PREFIX,
+  koboSourceDeviceKey,
+  koboStatisticsSessionId,
+  koboStatisticsSessionIdPrefix,
+} from './kobo-statistics-session.util';
+
+describe('koboStatisticsSessionId', () => {
+  it('carries the prefix the measured-session lookup excludes on', () => {
+    expect(koboStatisticsSessionId(30, 53, 0, 860).startsWith(KOBO_STATISTICS_SESSION_ID_PREFIX)).toBe(true);
+  });
+
+  it('resolves a replayed push to the same id', () => {
+    expect(koboStatisticsSessionId(30, 53, 0, 860)).toBe(koboStatisticsSessionId(30, 53, 0, 860));
+  });
+
+  it('separates devices, books, generations, and counter totals', () => {
+    const base = koboStatisticsSessionId(30, 53, 0, 860);
+    expect(koboStatisticsSessionId(31, 53, 0, 860)).not.toBe(base);
+    expect(koboStatisticsSessionId(30, 54, 0, 860)).not.toBe(base);
+    expect(koboStatisticsSessionId(30, 53, 1, 860)).not.toBe(base);
+    expect(koboStatisticsSessionId(30, 53, 0, 861)).not.toBe(base);
+  });
+
+  it('builds the device key and estimate prefix from the same device id', () => {
+    expect(koboSourceDeviceKey(30)).toBe('30');
+    expect(koboStatisticsSessionIdPrefix(30)).toBe('kst:30:');
+  });
+
+  it('stays inside the session id column', () => {
+    expect(koboStatisticsSessionId(2_147_483_647, 2_147_483_647, 2_147_483_647, 35_791_394).length).toBeLessThanOrEqual(64);
+  });
+});

server/src/modules/kobo/kobo-statistics-session.util.ts

@@ -0,0 +1,29 @@
+/**
+ * Identifiers for reading sessions derived from the reading counter a Kobo reports alongside
+ * its bookmark, rather than measured from the analytics events it no longer sends.
+ *
+ * A reading session carries no column saying where it came from beyond `source`, and `source`
+ * is `kobo` for both kinds. The prefix is what lets the derivation find, and exclude itself
+ * from, the measured sessions it must never compete with.
+ */
+export const KOBO_STATISTICS_SESSION_ID_PREFIX = 'kst:';
+export const KOBO_STATISTICS_CURSOR_SOURCE = 'kobo-statistics';
+
+export function koboSourceDeviceKey(deviceId: number): string {
+  return String(deviceId);
+}
+
+export function koboStatisticsSessionIdPrefix(deviceId: number): string {
+  return `${KOBO_STATISTICS_SESSION_ID_PREFIX}${deviceId}:`;
+}
+
+/**
+ * Derived rather than random, so a device replaying the same state push resolves to the id it
+ * already spent and is dropped instead of counted again.
+ *
+ * The device and cursor generation keep independent devices and re-reads apart. The cumulative
+ * counter makes retries within one generation idempotent.
+ */
+export function koboStatisticsSessionId(deviceId: number, bookFileId: number, generation: number, spentReadingMinutes: number): string {
+  return `${koboStatisticsSessionIdPrefix(deviceId)}${bookFileId}:${generation}:${spentReadingMinutes}`;
+}

server/src/modules/kobo/services/kobo-analytics.service.test.ts

@@ -10,6 +10,7 @@ describe('KoboAnalyticsService', () => {
   const bookService = { bulkSetRating: vi.fn() };
   const user = { id: 7 } as never;
   const device = { deviceId: 2, deviceToken: 'tok', userId: 7 } as never;
+  const syncOptions = { sourceDeviceKey: '2', estimateSessionIdPrefix: 'kst:2:' };
 
   function makeService() {
     return new KoboAnalyticsService(bookIdentityService as never, resolver as never, readingSessionService as never, bookService as never);
@@ -68,6 +69,7 @@ describe('KoboAnalyticsService', () => {
       },
       user,
       'kobo',
+      syncOptions,
     );
     expect(readingSessionService.save).toHaveBeenNthCalledWith(
       2,
@@ -79,6 +81,7 @@ describe('KoboAnalyticsService', () => {
       }),
       user,
       'kobo',
+      syncOptions,
     );
     expect(readingSessionService.save).toHaveBeenNthCalledWith(
       3,
@@ -89,6 +92,7 @@ describe('KoboAnalyticsService', () => {
       }),
       user,
       'kobo',
+      syncOptions,
     );
   });
 
@@ -109,7 +113,7 @@ describe('KoboAnalyticsService', () => {
       device,
     );
 
-    expect(readingSessionService.save).toHaveBeenCalledWith(100, expect.objectContaining({ sessionId: 'kobo-src' }), user, 'kobo');
+    expect(readingSessionService.save).toHaveBeenCalledWith(100, expect.objectContaining({ sessionId: 'kobo-src' }), user, 'kobo', syncOptions);
   });
 
   it('pairs OpenContent and LeaveContent to save progress delta and Kobo-reported duration', async () => {
@@ -164,6 +168,7 @@ describe('KoboAnalyticsService', () => {
       },
       user,
       'kobo',
+      syncOptions,
     );
     expect(readingSessionService.save).toHaveBeenNthCalledWith(
       2,
@@ -178,6 +183,7 @@ describe('KoboAnalyticsService', () => {
       },
       user,
       'kobo',
+      syncOptions,
     );
   });
 
@@ -223,6 +229,7 @@ describe('KoboAnalyticsService', () => {
       expect.objectContaining({ sessionId: 'leave-one', progressDelta: 2 }),
       user,
       'kobo',
+      syncOptions,
     );
     expect(readingSessionService.save).toHaveBeenNthCalledWith(
       2,
@@ -230,6 +237,7 @@ describe('KoboAnalyticsService', () => {
       expect.objectContaining({ sessionId: 'leave-two', progressDelta: 5 }),
       user,
       'kobo',
+      syncOptions,
     );
   });
 
@@ -261,6 +269,7 @@ describe('KoboAnalyticsService', () => {
       expect.objectContaining({ sessionId: 'leave-after-bad-progress', progressDelta: null, endProgress: 20 }),
       user,
       'kobo',
+      syncOptions,
     );
   });
 
@@ -300,6 +309,7 @@ describe('KoboAnalyticsService', () => {
       },
       user,
       'kobo',
+      syncOptions,
     );
   });
 
@@ -327,8 +337,8 @@ describe('KoboAnalyticsService', () => {
       device,
     );
 
-    expect(readingSessionService.save).toHaveBeenNthCalledWith(1, 100, expect.objectContaining({ endProgress: null }), user, 'kobo');
-    expect(readingSessionService.save).toHaveBeenNthCalledWith(2, 100, expect.objectContaining({ endProgress: null }), user, 'kobo');
+    expect(readingSessionService.save).toHaveBeenNthCalledWith(1, 100, expect.objectContaining({ endProgress: null }), user, 'kobo', syncOptions);
+    expect(readingSessionService.save).toHaveBeenNthCalledWith(2, 100, expect.objectContaining({ endProgress: null }), user, 'kobo', syncOptions);
   });
 
   it('still processes later events when an earlier save throws', async () => {
@@ -408,7 +418,13 @@ describe('KoboAnalyticsService', () => {
 
     expect(bookIdentityService.resolveBookIdByEntitlementId).toHaveBeenCalledWith(7, entitlementId);
     expect(resolver.resolveBookFileId).toHaveBeenCalledWith(7, 2, 9);
-    expect(readingSessionService.save).toHaveBeenCalledWith(100, expect.objectContaining({ sessionId: 'uuid-volume', endProgress: 8 }), user, 'kobo');
+    expect(readingSessionService.save).toHaveBeenCalledWith(
+      100,
+      expect.objectContaining({ sessionId: 'uuid-volume', endProgress: 8 }),
+      user,
+      'kobo',
+      syncOptions,
+    );
   });
 
   it('skips LeaveContent without volumeid or SecondsRead', async () => {

server/src/modules/kobo/services/kobo-analytics.service.ts

@@ -6,6 +6,7 @@ import { BookService } from '../../book/book.service';
 import { ReadingSessionService } from '../../reading-session/reading-session.service';
 import type { KoboDeviceContext } from '../guards/kobo-token.guard';
 import type { KoboAnalyticsBody, KoboAnalyticsEvent } from '../kobo-analytics.types';
+import { koboSourceDeviceKey, koboStatisticsSessionIdPrefix } from '../kobo-statistics-session.util';
 import { KoboBookIdentityService } from './kobo-book-identity.service';
 import { KoboAnalyticsResolverService } from './kobo-analytics-resolver.service';
 
@@ -178,6 +179,10 @@ export class KoboAnalyticsService {
       },
       user,
       'kobo',
+      {
+        sourceDeviceKey: koboSourceDeviceKey(deviceId),
+        estimateSessionIdPrefix: koboStatisticsSessionIdPrefix(deviceId),
+      },
     );
   }
 

server/src/modules/kobo/services/kobo-reading-state.service.test.ts

@@ -63,6 +63,8 @@ describe('KoboReadingStateService', () => {
   const progressBridge = { koboBookmarkToCanonical: vi.fn(), cfiToKoboBookmark: vi.fn() };
   const settingsService = { getSettings: vi.fn() };
   const achievementEvents = { emit: vi.fn() };
+  const analyticsResolver = { resolveBookFileId: vi.fn() };
+  const readingSessions = { recordCumulativeSyncedSession: vi.fn() };
 
   function makeService(db: ReturnType<typeof makeDb>) {
     return new KoboReadingStateService(
@@ -73,6 +75,8 @@ describe('KoboReadingStateService', () => {
       progressBridge as never,
       settingsService as never,
       achievementEvents as never,
+      analyticsResolver as never,
+      readingSessions as never,
     );
   }
 
@@ -83,6 +87,8 @@ describe('KoboReadingStateService', () => {
     progressBridge.koboBookmarkToCanonical.mockResolvedValue(null);
     progressBridge.cfiToKoboBookmark.mockResolvedValue(null);
     settingsService.getSettings.mockResolvedValue({ twoWayProgressSync: false });
+    analyticsResolver.resolveBookFileId.mockResolvedValue({ kind: 'resolved', bookFileId: 53 });
+    readingSessions.recordCumulativeSyncedSession.mockResolvedValue({ kind: 'baseline' });
     bookIdentityService.ensureForBook.mockImplementation((_userId: number, bookId: number) => ({
       bookId,
       entitlementId: `entitlement-${bookId}`,
@@ -219,7 +225,7 @@ describe('KoboReadingStateService', () => {
     expect(achievementEvents.emit).not.toHaveBeenCalled();
   });
 
-  it('does not propagate or repeat side effects for an identical device echo', async () => {
+  it('does not repeat bookmark or status side effects for an identical device echo', async () => {
     const db = makeDb();
     const stateInsert = makeInsertChain();
     const bookmark = { LastModified: '2026-01-02T00:00:00.000Z', ProgressPercent: 42 };
@@ -254,7 +260,7 @@ describe('KoboReadingStateService', () => {
     );
 
     expect(db.execute).not.toHaveBeenCalled();
-    expect(db.select).not.toHaveBeenCalled();
+    expect(readingSessions.recordCumulativeSyncedSession).toHaveBeenCalledWith(expect.objectContaining({ counter: 5, sourceDeviceKey: '30' }));
     expect(progressBridge.koboBookmarkToCanonical).not.toHaveBeenCalled();
     expect(userBookStatusService.autoUpdate).not.toHaveBeenCalled();
     expect(achievementEvents.emit).not.toHaveBeenCalled();
@@ -631,4 +637,106 @@ describe('KoboReadingStateService', () => {
     expect(queries[0]!.sql).toContain('"updated_at" = "reading_progress"."updated_at"');
     expect(queries[0]!.sql).toContain('"last_read_at" = "reading_progress"."last_read_at"');
   });
+  describe('reading sessions derived from device reading time', () => {
+    const STATS_AT = '2026-08-20T19:30:00.000Z';
+
+    function makeStatsDb(previousStatistics: unknown, settings?: unknown) {
+      const db = makeDb();
+      db.insert.mockReturnValue(makeInsertChain());
+      db.query.books.findFirst.mockResolvedValue({ id: 46 });
+      db.query.koboReadingStates.findFirst.mockResolvedValue(previousStatistics === undefined ? null : { statistics: previousStatistics });
+      db.select.mockImplementation((columns: Record<string, unknown>) => {
+        const selection = Object.keys(columns ?? {}).join(',');
+        if (selection === 'settings') return makeSelectChain(settings === undefined ? [] : [{ settings }]);
+        return makeSelectChain([]);
+      });
+      return db;
+    }
+
+    function push(db: ReturnType<typeof makeDb>, statistics: Record<string, unknown>, extra: Record<string, unknown> = {}) {
+      return makeService(db).upsertState(7, 46, { Statistics: statistics, ...extra }, 1, 99, false, 30);
+    }
+
+    it('passes the raw device counter to the per-device transactional recorder', async () => {
+      const db = makeStatsDb({ SpentReadingMinutes: 800 });
+
+      await push(db, { LastModified: STATS_AT, SpentReadingMinutes: 860 });
+
+      expect(readingSessions.recordCumulativeSyncedSession).toHaveBeenCalledWith(
+        expect.objectContaining({
+          userId: 7,
+          bookId: 46,
+          bookFileId: 53,
+          cursorSource: 'kobo-statistics',
+          sourceDeviceKey: '30',
+          sessionIdPrefix: 'kst:30:',
+          counter: 860,
+          endedAt: new Date(STATS_AT),
+          source: 'kobo',
+        }),
+      );
+    });
+
+    it('processes an identical global state so a second device can establish its own cursor', async () => {
+      const statistics = { LastModified: STATS_AT, SpentReadingMinutes: 860 };
+      const db = makeStatsDb(statistics);
+
+      await push(db, { ...statistics });
+
+      expect(readingSessions.recordCumulativeSyncedSession).toHaveBeenCalledWith(expect.objectContaining({ sourceDeviceKey: '30', counter: 860 }));
+    });
+
+    it('keeps the cursor retryable when the file cannot currently be resolved', async () => {
+      const db = makeStatsDb({ SpentReadingMinutes: 800 });
+      analyticsResolver.resolveBookFileId.mockResolvedValue({ kind: 'skipped', reason: 'no_epub_file' });
+
+      await push(db, { LastModified: STATS_AT, SpentReadingMinutes: 860 });
+
+      expect(readingSessions.recordCumulativeSyncedSession).toHaveBeenCalledWith(expect.objectContaining({ bookFileId: null, counter: 860 }));
+    });
+
+    it('carries progress and the reporting user timezone', async () => {
+      const db = makeStatsDb({ SpentReadingMinutes: 800 }, { timezone: 'Europe/Oslo' });
+      db.query.koboReadingStates.findFirst.mockResolvedValue({
+        statistics: { SpentReadingMinutes: 800 },
+        currentBookmark: { LastModified: '2026-08-19T10:00:00.000Z', ProgressPercent: 22 },
+      });
+
+      await push(db, { LastModified: STATS_AT, SpentReadingMinutes: 860 }, { CurrentBookmark: { LastModified: STATS_AT, ProgressPercent: 29 } });
+
+      expect(readingSessions.recordCumulativeSyncedSession).toHaveBeenCalledWith(
+        expect.objectContaining({ progressDelta: 7, endProgress: 29, timeZone: 'Europe/Oslo' }),
+      );
+    });
+
+    it('holds a device clock running ahead to the present', async () => {
+      const db = makeStatsDb({ SpentReadingMinutes: 800 });
+      const before = Date.now();
+
+      await push(db, { LastModified: '2099-01-01T00:00:00.000Z', SpentReadingMinutes: 860 });
+
+      const call = readingSessions.recordCumulativeSyncedSession.mock.calls[0]![0] as { endedAt: Date };
+      expect(call.endedAt.getTime()).toBeGreaterThanOrEqual(before);
+      expect(call.endedAt.getTime()).toBeLessThanOrEqual(Date.now());
+    });
+
+    it('rejects a counter too large for the reading-session duration column', async () => {
+      const db = makeStatsDb({ SpentReadingMinutes: 1 });
+
+      await push(db, { LastModified: STATS_AT, SpentReadingMinutes: 40_000_000 });
+
+      expect(readingSessions.recordCumulativeSyncedSession).not.toHaveBeenCalled();
+    });
+
+    it('still acknowledges the push when transactional accounting fails', async () => {
+      const db = makeStatsDb({ SpentReadingMinutes: 800 });
+      readingSessions.recordCumulativeSyncedSession.mockRejectedValue(new Error('database is down'));
+      const warn = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
+
+      await expect(push(db, { LastModified: STATS_AT, SpentReadingMinutes: 860 })).resolves.toEqual(ACK('entitlement-46', 'Success'));
+
+      expect(warn).toHaveBeenCalledWith(expect.stringContaining('[kobo.statistics_session] [fail]'));
+      warn.mockRestore();
+    });
+  });
 });

server/src/modules/kobo/services/kobo-reading-state.service.ts

@@ -6,17 +6,28 @@ import { NodePgDatabase } from 'drizzle-orm/node-postgres';
 import { DB } from '../../../db';
 import * as schema from '../../../db/schema';
 import { UserBookStatusService } from '../../user-book-status/user-book-status.service';
+import { ReadingSessionService } from '../../reading-session/reading-session.service';
 import { AchievementEventsService, ACHIEVEMENT_EVENT_BOOK_PROGRESS_CHANGED } from '../../achievement/achievement-events.service';
+import {
+  KOBO_STATISTICS_CURSOR_SOURCE,
+  koboSourceDeviceKey,
+  koboStatisticsSessionId,
+  koboStatisticsSessionIdPrefix,
+} from '../kobo-statistics-session.util';
+import { KoboAnalyticsResolverService } from './kobo-analytics-resolver.service';
 import { KoboBookAccessService } from './kobo-book-access.service';
 import { KoboBookIdentityService } from './kobo-book-identity.service';
 import { KoboProgressBridgeService } from './kobo-progress-bridge.service';
 import { KoboSettingsService } from './kobo-settings.service';
 import { advanceIsoTimestamp, maxIsoTimestamp } from '../../../common/utils/iso-timestamp.utils';
 import { sanitizeLogValue } from '../../../common/utils/log-sanitize.utils';
+import { resolveTimeZone } from '../../../common/utils/timezone.utils';
 
 type Db = NodePgDatabase<typeof schema>;
 type JsonObj = Record<string, unknown>;
 const PROGRESS_EPSILON = 0.0001;
+const STATISTICS_SESSION_EVENT = 'kobo.statistics_session';
+const MAX_STATISTICS_MINUTES = Math.floor(2_147_483_647 / 60);
 
 type KoboSectionResult = { Result: 'Success' | 'Ignored' };
 
@@ -76,6 +87,8 @@ export class KoboReadingStateService {
     private readonly progressBridge: KoboProgressBridgeService,
     private readonly settingsService: KoboSettingsService,
     private readonly achievementEvents: AchievementEventsService,
+    private readonly analyticsResolver: KoboAnalyticsResolverService,
+    private readonly readingSessions: ReadingSessionService,
   ) {}
 
   async upsertState(
@@ -113,6 +126,7 @@ export class KoboReadingStateService {
     });
 
     const previousBookmark = this.asJsonObj(existing?.currentBookmark ?? null);
+    const previousStats = this.asJsonObj(existing?.statistics ?? null);
     const previousStatus = this.asJsonObj(existing?.statusInfo ?? null);
     const previousPercent = this.extractPercent(previousBookmark);
     const previousTimesStarted = typeof previousStatus?.TimesStartedReading === 'number' ? previousStatus.TimesStartedReading : null;
@@ -121,7 +135,7 @@ export class KoboReadingStateService {
     const mergedStats = mergeSubObject(incomingStats, existing?.statistics as JsonObj | null);
     const mergedStatus = mergeSubObject(incomingStatus, existing?.statusInfo as JsonObj | null);
     const bookmarkChanged = !isDeepStrictEqual(mergedBookmark, previousBookmark);
-    const statisticsChanged = !isDeepStrictEqual(mergedStats, this.asJsonObj(existing?.statistics ?? null));
+    const statisticsChanged = !isDeepStrictEqual(mergedStats, previousStats);
     const statusChanged = !isDeepStrictEqual(mergedStatus, previousStatus);
     const stateChanged = bookmarkChanged || statisticsChanged || statusChanged;
 
@@ -198,6 +212,20 @@ export class KoboReadingStateService {
       });
     }
 
+    // After the status update, so the attempt a first push opens is already there for the
+    // session to be filed against rather than landing with a null attemptId.
+    if (incomingStats) {
+      await this.recordStatisticsReadingSession({
+        userId,
+        bookId,
+        sourceDeviceId,
+        incomingStats,
+        previousPercent,
+        mergedPercent,
+        fallbackLastModified: lastModified,
+      });
+    }
+
     if (bookmarkChanged && mergedPercent !== null) {
       this.achievementEvents.emit(ACHIEVEMENT_EVENT_BOOK_PROGRESS_CHANGED, {
         userId,
@@ -210,6 +238,93 @@ export class KoboReadingStateService {
     return buildStateUpdateResponse(entitlementId, 'Success');
   }
 
+  /**
+   * Turns the reading time a device reports alongside its bookmark into a reading session.
+   *
+   * Current Kobo firmware no longer emits the `LeaveContent` analytics events the reading log
+   * was built on, so `POST /v1/analytics/event` never arrives and every Kobo statistic reads
+   * zero however much the device is read. What the device does still send, on every state push,
+   * is its own `SpentReadingMinutes` - the counter it keeps as `content.TimeSpentReading` on
+   * device, and the only reading time these devices offer at all.
+   *
+   * Each device has an independent durable cursor. Session insertion, daily aggregation, and
+   * cursor advancement share one transaction so a failed write is retried from the same counter.
+   * Measured analytics sessions and counter-derived sessions are serialized by the same lock and
+   * reconciled by overlap, regardless of which endpoint arrives first.
+   *
+   * Failures are swallowed. A device retries the entire push when the response is not a clean
+   * acknowledgement, so a session that cannot be stored must not take the bookmark down with it.
+   */
+  private async recordStatisticsReadingSession(params: {
+    userId: number;
+    bookId: number;
+    sourceDeviceId: number;
+    incomingStats: JsonObj;
+    previousPercent: number | null;
+    mergedPercent: number | null;
+    fallbackLastModified: string;
+  }): Promise<void> {
+    const { userId, bookId, sourceDeviceId, incomingStats, previousPercent, mergedPercent } = params;
+
+    const currentMinutes = this.extractSpentReadingMinutes(incomingStats);
+    if (currentMinutes === null) return;
+
+    const startedAtMs = Date.now();
+    try {
+      const resolved = await this.analyticsResolver.resolveBookFileId(userId, sourceDeviceId, bookId);
+      const endedAt = this.resolveStatisticsEndedAt(incomingStats, params.fallbackLastModified);
+      const result = await this.readingSessions.recordCumulativeSyncedSession({
+        userId,
+        bookId,
+        bookFileId: resolved.kind === 'resolved' ? resolved.bookFileId : null,
+        cursorSource: KOBO_STATISTICS_CURSOR_SOURCE,
+        sourceDeviceKey: koboSourceDeviceKey(sourceDeviceId),
+        sessionIdPrefix: koboStatisticsSessionIdPrefix(sourceDeviceId),
+        buildSessionId: (bookFileId, generation, counter) => koboStatisticsSessionId(sourceDeviceId, bookFileId, generation, counter),
+        counter: currentMinutes,
+        endedAt,
+        progressDelta: this.resolveProgressDelta(previousPercent, mergedPercent),
+        endProgress: mergedPercent,
+        source: 'kobo',
+        timeZone: await this.findUserTimeZone(userId),
+      });
+
+      this.logger.log(
+        `[${STATISTICS_SESSION_EVENT}] [end] userId=${userId} bookId=${bookId} deviceId=${sourceDeviceId} bookFileId=${resolved.kind === 'resolved' ? resolved.bookFileId : 'none'} durationMs=${Date.now() - startedAtMs} spentReadingMinutes=${currentMinutes} outcome=${result.kind}${result.kind === 'skipped' ? ` reason=${result.reason}` : ''} - device reading-time counter processed`,
+      );
+    } catch (error: unknown) {
+      const err = error instanceof Error ? error : new Error(String(error));
+      this.logger.warn(
+        `[${STATISTICS_SESSION_EVENT}] [fail] userId=${userId} bookId=${bookId} durationMs=${Date.now() - startedAtMs} errorClass=${err.constructor.name} error="${sanitizeLogValue(err.message)}" - deriving a reading session from device reading time failed`,
+      );
+    }
+  }
+
+  /** Whole minutes only, so a fractional counter accrues across pushes rather than truncating on each. */
+  private extractSpentReadingMinutes(stats: JsonObj | null): number | null {
+    const value = stats?.SpentReadingMinutes;
+    if (typeof value !== 'number' || !Number.isFinite(value) || value < 0 || value > MAX_STATISTICS_MINUTES) return null;
+    return Math.floor(value);
+  }
+
+  /** The device's own clock for the counter, held to the present so a fast clock cannot book time into tomorrow. */
+  private resolveStatisticsEndedAt(stats: JsonObj | null, fallback: string): Date {
+    const modified = typeof stats?.LastModified === 'string' ? stats.LastModified : undefined;
+    const parsed = this.parseKoboTimestamp(modified) ?? this.parseKoboTimestamp(fallback) ?? new Date();
+    const now = new Date();
+    return parsed.getTime() > now.getTime() ? now : parsed;
+  }
+
+  private resolveProgressDelta(previousPercent: number | null, mergedPercent: number | null): number | null {
+    if (previousPercent === null || mergedPercent === null) return null;
+    return Math.round(Math.max(-100, Math.min(100, mergedPercent - previousPercent)) * 100) / 100;
+  }
+
+  private async findUserTimeZone(userId: number): Promise<string> {
+    const [row] = await this.db.select({ settings: schema.users.settings }).from(schema.users).where(eq(schema.users.id, userId)).limit(1);
+    return resolveTimeZone((row?.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC');
+  }
+
   private async autoUpdateReadStatus(
     userId: number,
     bookId: number,

server/src/modules/reading-session/reading-session-cumulative.repository.test.ts

@@ -0,0 +1,241 @@
+import { bookFiles, readingSessionSyncCursors, readingSessions, userReadingDailyStats } from '../../db/schema';
+import { ReadingSessionRepository, type RecordCumulativeReadingSessionParams } from './reading-session.repository';
+
+function makeLimitChain(rows: unknown[]) {
+  const limit = vi.fn().mockResolvedValue(rows);
+  const where = vi.fn().mockReturnValue({ limit });
+  return { where, limit };
+}
+
+function makeHarness(
+  options: {
+    cursor?: { counter: number; generation: number; lastModified: Date } | null;
+    measured?: boolean;
+    file?: { bookId: number; libraryId: number } | null;
+    inserted?: boolean;
+  } = {},
+) {
+  const cursorChain = makeLimitChain(options.cursor ? [options.cursor] : []);
+  const measuredChain = makeLimitChain(options.measured ? [{ id: 88 }] : []);
+  const fileChain = makeLimitChain(options.file === null ? [] : [options.file ?? { bookId: 46, libraryId: 9 }]);
+
+  const select = vi.fn().mockImplementation(() => ({
+    from: vi.fn().mockImplementation((table: unknown) => {
+      if (table === readingSessionSyncCursors) return { where: cursorChain.where };
+      if (table === readingSessions) return { where: measuredChain.where };
+      if (table === bookFiles) return { innerJoin: vi.fn().mockReturnValue({ where: fileChain.where }) };
+      throw new Error('Unexpected table in select');
+    }),
+  }));
+
+  const cursorInsertValues = vi.fn().mockResolvedValue(undefined);
+  const sessionReturning = vi.fn().mockResolvedValue(options.inserted === false ? [] : [{ id: 101 }]);
+  const sessionConflict = vi.fn().mockReturnValue({ returning: sessionReturning });
+  const sessionValues = vi.fn().mockReturnValue({ onConflictDoNothing: sessionConflict });
+  const dailyValues = vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) });
+  const insert = vi.fn().mockImplementation((table: unknown) => {
+    if (table === readingSessionSyncCursors) return { values: cursorInsertValues };
+    if (table === readingSessions) return { values: sessionValues };
+    if (table === userReadingDailyStats) return { values: dailyValues };
+    throw new Error('Unexpected table in insert');
+  });
+
+  const cursorUpdateWhere = vi.fn().mockResolvedValue(undefined);
+  const cursorUpdateSet = vi.fn().mockReturnValue({ where: cursorUpdateWhere });
+  const update = vi.fn().mockReturnValue({ set: cursorUpdateSet });
+  const tx = { execute: vi.fn().mockResolvedValue(undefined), select, insert, update };
+  const transaction = vi.fn(async (callback: (trx: typeof tx) => Promise<unknown>) => callback(tx));
+  const repo = new ReadingSessionRepository({ transaction } as never);
+
+  return { repo, tx, transaction, cursorInsertValues, cursorUpdateSet, sessionValues, dailyValues };
+}
+
+const params = (overrides: Partial<RecordCumulativeReadingSessionParams> = {}): RecordCumulativeReadingSessionParams => ({
+  userId: 7,
+  bookId: 46,
+  bookFileId: 53,
+  cursorSource: 'kobo-statistics',
+  sourceDeviceKey: '30',
+  sessionIdPrefix: 'kst:30:',
+  buildSessionId: (bookFileId, generation, counter) => `kst:30:${bookFileId}:${generation}:${counter}`,
+  counter: 860,
+  endedAt: new Date('2026-08-20T19:30:00.000Z'),
+  progressDelta: 7,
+  endProgress: 29,
+  source: 'kobo',
+  timeZone: 'Europe/Oslo',
+  ...overrides,
+});
+
+describe('ReadingSessionRepository.recordCumulativeSyncedSession', () => {
+  it('establishes an independent device baseline without creating a session', async () => {
+    const { repo, cursorInsertValues, sessionValues } = makeHarness();
+
+    await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({ kind: 'baseline' });
+
+    expect(cursorInsertValues).toHaveBeenCalledWith(expect.objectContaining({ userId: 7, bookId: 46, sourceDeviceKey: '30', counter: 860 }));
+    expect(sessionValues).not.toHaveBeenCalled();
+  });
+
+  it('atomically stores the device delta and advances its cursor', async () => {
+    const { repo, transaction, cursorUpdateSet, sessionValues, dailyValues } = makeHarness({
+      cursor: { counter: 800, generation: 2, lastModified: new Date('2026-08-19T10:00:00.000Z') },
+    });
+
+    await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({
+      kind: 'saved',
+      durationSeconds: 3600,
+      sessionId: 'kst:30:53:2:860',
+    });
+
+    expect(transaction).toHaveBeenCalledOnce();
+    expect(sessionValues).toHaveBeenCalledWith(
+      expect.objectContaining({
+        userId: 7,
+        bookId: 46,
+        bookFileId: 53,
+        sourceDeviceKey: '30',
+        sessionId: 'kst:30:53:2:860',
+        durationSeconds: 3600,
+      }),
+    );
+    expect(cursorUpdateSet).toHaveBeenCalledWith(expect.objectContaining({ counter: 860, generation: 2 }));
+    expect(dailyValues).toHaveBeenCalledOnce();
+  });
+
+  it('does not advance the cursor when the file is temporarily unavailable', async () => {
+    const { repo, cursorUpdateSet, sessionValues } = makeHarness({
+      cursor: { counter: 800, generation: 0, lastModified: new Date('2026-08-19T10:00:00.000Z') },
+    });
+
+    await expect(repo.recordCumulativeSyncedSession(params({ bookFileId: null }))).resolves.toEqual({
+      kind: 'skipped',
+      reason: 'book_file_not_found',
+    });
+
+    expect(cursorUpdateSet).not.toHaveBeenCalled();
+    expect(sessionValues).not.toHaveBeenCalled();
+  });
+
+  it('advances without an estimate when overlapping measured analytics already exist', async () => {
+    const { repo, cursorUpdateSet, sessionValues } = makeHarness({
+      cursor: { counter: 800, generation: 0, lastModified: new Date('2026-08-19T10:00:00.000Z') },
+      measured: true,
+    });
+
+    await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({
+      kind: 'skipped',
+      reason: 'measured_session_present',
+    });
+
+    expect(cursorUpdateSet).toHaveBeenCalledWith(expect.objectContaining({ counter: 860 }));
+    expect(sessionValues).not.toHaveBeenCalled();
+  });
+
+  it('starts a new generation when a newer device counter resets', async () => {
+    const { repo, cursorUpdateSet, sessionValues } = makeHarness({
+      cursor: { counter: 860, generation: 4, lastModified: new Date('2026-08-19T10:00:00.000Z') },
+    });
+
+    await expect(repo.recordCumulativeSyncedSession(params({ counter: 0 }))).resolves.toEqual({ kind: 'reset' });
+
+    expect(cursorUpdateSet).toHaveBeenCalledWith(expect.objectContaining({ counter: 0, generation: 5 }));
+    expect(sessionValues).not.toHaveBeenCalled();
+  });
+
+  it('does not let an out-of-order push rewind a device cursor', async () => {
+    const { repo, cursorUpdateSet } = makeHarness({
+      cursor: { counter: 860, generation: 1, lastModified: new Date('2026-08-21T10:00:00.000Z') },
+    });
+
+    await expect(repo.recordCumulativeSyncedSession(params({ counter: 900 }))).resolves.toEqual({ kind: 'stale' });
+    expect(cursorUpdateSet).not.toHaveBeenCalled();
+  });
+
+  it('advances the cursor after an idempotent duplicate', async () => {
+    const { repo, cursorUpdateSet } = makeHarness({
+      cursor: { counter: 800, generation: 0, lastModified: new Date('2026-08-19T10:00:00.000Z') },
+      inserted: false,
+    });
+
+    await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({
+      kind: 'skipped',
+      reason: 'duplicate_session_id',
+    });
+    expect(cursorUpdateSet).toHaveBeenCalledWith(expect.objectContaining({ counter: 860 }));
+  });
+});
+
+describe('ReadingSessionRepository measured-session reconciliation', () => {
+  it('stores the device key and removes an overlapping estimate in the same transaction', async () => {
+    const outerLimit = vi.fn().mockResolvedValue([{ bookId: 46, libraryId: 9 }]);
+    const outerSelect = vi.fn().mockReturnValue({
+      from: vi.fn().mockReturnValue({ innerJoin: vi.fn().mockReturnValue({ where: vi.fn().mockReturnValue({ limit: outerLimit }) }) }),
+    });
+
+    const estimate = {
+      id: 20,
+      startedAt: new Date('2026-08-20T18:30:00.000Z'),
+      endedAt: new Date('2026-08-20T19:30:00.000Z'),
+      durationSeconds: 3600,
+      progressDelta: 7,
+    };
+    const estimateSelect = {
+      from: vi.fn().mockReturnValue({
+        where: vi.fn().mockReturnValue({ orderBy: vi.fn().mockReturnValue({ limit: vi.fn().mockResolvedValue([estimate]) }) }),
+      }),
+    };
+    const survivingSelect = {
+      from: vi.fn().mockReturnValue({
+        innerJoin: vi.fn().mockReturnValue({
+          where: vi.fn().mockResolvedValue([
+            {
+              startedAt: new Date('2026-08-20T19:20:00.000Z'),
+              endedAt: new Date('2026-08-20T19:30:00.000Z'),
+              durationSeconds: 600,
+              progressDelta: 2,
+            },
+          ]),
+        }),
+      }),
+    };
+    const txSelect = vi.fn().mockReturnValueOnce(estimateSelect).mockReturnValueOnce(survivingSelect);
+
+    const sessionValues = vi.fn().mockReturnValue({
+      onConflictDoNothing: vi.fn().mockReturnValue({ returning: vi.fn().mockResolvedValue([{ id: 21 }]) }),
+    });
+    const dailyValues = vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) });
+    const insert = vi.fn().mockImplementation((table: unknown) => {
+      if (table === readingSessions) return { values: sessionValues };
+      if (table === userReadingDailyStats) return { values: dailyValues };
+      throw new Error('Unexpected table in insert');
+    });
+    const deleteFn = vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) });
+    const tx = { execute: vi.fn().mockResolvedValue(undefined), insert, select: txSelect, delete: deleteFn };
+    const transaction = vi.fn(async (callback: (trx: typeof tx) => Promise<unknown>) => callback(tx));
+    const repo = new ReadingSessionRepository({ select: outerSelect, transaction } as never);
+
+    await expect(
+      repo.saveSession(
+        7,
+        53,
+        'measured-event',
+        new Date('2026-08-20T19:20:00.000Z'),
+        new Date('2026-08-20T19:30:00.000Z'),
+        600,
+        2,
+        29,
+        'kobo',
+        'UTC',
+        { sourceDeviceKey: '30', estimateSessionIdPrefix: 'kst:30:' },
+      ),
+    ).resolves.toEqual({ kind: 'saved' });
+
+    expect(transaction).toHaveBeenCalledOnce();
+    expect(sessionValues).toHaveBeenCalledWith(expect.objectContaining({ sessionId: 'measured-event', sourceDeviceKey: '30' }));
+    expect(deleteFn).toHaveBeenCalledWith(readingSessions);
+    expect(dailyValues).toHaveBeenLastCalledWith([
+      expect.objectContaining({ day: '2026-08-20', readingSeconds: 600, progressDelta: 2, sessionsCount: 1 }),
+    ]);
+  });
+});

server/src/modules/reading-session/reading-session.repository.ts

@@ -1,5 +1,5 @@
 import { Inject, Injectable } from '@nestjs/common';
-import { and, asc, count, desc, eq, gt, gte, inArray, isNotNull, like, lt, lte, max, min, sql } from 'drizzle-orm';
+import { and, asc, count, desc, eq, gt, gte, inArray, isNotNull, isNull, like, lt, lte, max, min, notLike, sql } from 'drizzle-orm';
 import type { NodePgDatabase } from 'drizzle-orm/node-postgres';
 
 import type {
@@ -20,7 +20,7 @@ import {
 import { toDateKeyInTimeZone } from '../../common/utils/timezone.utils';
 import { DB } from '../../db';
 import * as schema from '../../db/schema';
-import { bookFiles, books, readingSessions, userReadingDailyStats } from '../../db/schema';
+import { bookFiles, books, readingSessionSyncCursors, readingSessions, userReadingDailyStats } from '../../db/schema';
 
 type Db = NodePgDatabase<typeof schema>;
 type Tx = Parameters<Parameters<Db['transaction']>[0]>[0];
@@ -35,6 +35,35 @@ export type SaveReadingSessionResult =
       reason: 'duration_below_minimum' | 'book_file_not_found' | 'duplicate_session_id';
     };
 
+export type RecordCumulativeReadingSessionResult =
+  | { kind: 'baseline' | 'unchanged' | 'reset' | 'stale' }
+  | { kind: 'saved'; durationSeconds: number; sessionId: string }
+  | {
+      kind: 'skipped';
+      reason: 'book_file_not_found' | 'duplicate_session_id' | 'measured_session_present';
+    };
+
+export interface ReadingSessionSyncOptions {
+  sourceDeviceKey: string;
+  estimateSessionIdPrefix?: string;
+}
+
+export interface RecordCumulativeReadingSessionParams {
+  userId: number;
+  bookId: number;
+  bookFileId: number | null;
+  cursorSource: string;
+  sourceDeviceKey: string;
+  sessionIdPrefix: string;
+  buildSessionId: (bookFileId: number, generation: number, counter: number) => string;
+  counter: number;
+  endedAt: Date;
+  progressDelta: number | null;
+  endProgress: number | null;
+  source: ReadingSessionSource;
+  timeZone: string;
+}
+
 export interface InsertManualSessionParams {
   userId: number;
   bookId: number;
@@ -64,6 +93,7 @@ export class ReadingSessionRepository {
     endProgress: number | null,
     source: ReadingSessionSource = 'web',
     timeZone = 'UTC',
+    sync?: ReadingSessionSyncOptions,
   ): Promise<SaveReadingSessionResult> {
     if (durationSeconds < MIN_READING_SESSION_SECONDS) {
       return { kind: 'skipped', reason: 'duration_below_minimum' };
@@ -83,6 +113,8 @@ export class ReadingSessionRepository {
     const { bookId, libraryId } = fileRow;
 
     return this.db.transaction(async (tx): Promise<SaveReadingSessionResult> => {
+      if (sync) await this.lockSyncSource(tx, userId, bookId, sync.sourceDeviceKey);
+
       const inserted = await tx
         .insert(readingSessions)
         .values({
@@ -92,6 +124,7 @@ export class ReadingSessionRepository {
           attemptId: sql`(select id from reading_attempts where user_id = ${userId} and book_id = ${bookId} and outcome is null and deleted_at is null limit 1)`,
           sessionId,
           source,
+          sourceDeviceKey: sync?.sourceDeviceKey ?? null,
           startedAt,
           endedAt,
           durationSeconds,
@@ -101,13 +134,159 @@ export class ReadingSessionRepository {
         .onConflictDoNothing({ target: [readingSessions.userId, readingSessions.sessionId] })
         .returning({ id: readingSessions.id });
 
-      if (inserted.length === 0) {
-        return { kind: 'skipped', reason: 'duplicate_session_id' };
+      const result: SaveReadingSessionResult = inserted.length === 0 ? { kind: 'skipped', reason: 'duplicate_session_id' } : { kind: 'saved' };
+
+      if (result.kind === 'saved') {
+        await this.upsertDailyStats(tx, { userId, libraryId, startedAt, endedAt, durationSeconds, progressDelta, timeZone });
+      } else if (sync) {
+        await tx
+          .update(readingSessions)
+          .set({ sourceDeviceKey: sync.sourceDeviceKey })
+          .where(
+            and(
+              eq(readingSessions.userId, userId),
+              eq(readingSessions.bookId, bookId),
+              eq(readingSessions.sessionId, sessionId),
+              eq(readingSessions.source, source),
+              isNull(readingSessions.sourceDeviceKey),
+            ),
+          );
       }
 
-      await this.upsertDailyStats(tx, { userId, libraryId, startedAt, endedAt, durationSeconds, progressDelta, timeZone });
+      if (sync?.estimateSessionIdPrefix) {
+        await this.deleteOverlappingSyncEstimates(tx, {
+          userId,
+          bookId,
+          libraryId,
+          source,
+          sourceDeviceKey: sync.sourceDeviceKey,
+          estimateSessionIdPrefix: sync.estimateSessionIdPrefix,
+          startedAt,
+          endedAt,
+          timeZone,
+        });
+      }
 
-      return { kind: 'saved' };
+      return result;
+    });
+  }
+
+  async recordCumulativeSyncedSession(params: RecordCumulativeReadingSessionParams): Promise<RecordCumulativeReadingSessionResult> {
+    return this.db.transaction(async (tx): Promise<RecordCumulativeReadingSessionResult> => {
+      await this.lockSyncSource(tx, params.userId, params.bookId, params.sourceDeviceKey);
+
+      const [cursor] = await tx
+        .select({
+          counter: readingSessionSyncCursors.counter,
+          generation: readingSessionSyncCursors.generation,
+          lastModified: readingSessionSyncCursors.lastModified,
+        })
+        .from(readingSessionSyncCursors)
+        .where(
+          and(
+            eq(readingSessionSyncCursors.userId, params.userId),
+            eq(readingSessionSyncCursors.bookId, params.bookId),
+            eq(readingSessionSyncCursors.source, params.cursorSource),
+            eq(readingSessionSyncCursors.sourceDeviceKey, params.sourceDeviceKey),
+          ),
+        )
+        .limit(1);
+
+      if (!cursor) {
+        await tx.insert(readingSessionSyncCursors).values({
+          userId: params.userId,
+          bookId: params.bookId,
+          source: params.cursorSource,
+          sourceDeviceKey: params.sourceDeviceKey,
+          counter: params.counter,
+          lastModified: params.endedAt,
+        });
+        return { kind: 'baseline' };
+      }
+
+      if (params.endedAt.getTime() < cursor.lastModified.getTime()) return { kind: 'stale' };
+
+      if (params.counter === cursor.counter) {
+        if (params.endedAt.getTime() > cursor.lastModified.getTime()) {
+          await this.updateSyncCursor(tx, params, cursor.generation);
+        }
+        return { kind: 'unchanged' };
+      }
+
+      if (params.counter < cursor.counter) {
+        await this.updateSyncCursor(tx, params, cursor.generation + 1);
+        return { kind: 'reset' };
+      }
+
+      const durationSeconds = (params.counter - cursor.counter) * 60;
+      const startedAt = new Date(params.endedAt.getTime() - durationSeconds * 1000);
+
+      const [measured] = await tx
+        .select({ id: readingSessions.id })
+        .from(readingSessions)
+        .where(
+          and(
+            eq(readingSessions.userId, params.userId),
+            eq(readingSessions.bookId, params.bookId),
+            eq(readingSessions.source, params.source),
+            eq(readingSessions.sourceDeviceKey, params.sourceDeviceKey),
+            notLike(readingSessions.sessionId, `${params.sessionIdPrefix}%`),
+            lt(readingSessions.startedAt, params.endedAt),
+            gt(readingSessions.endedAt, startedAt),
+          ),
+        )
+        .limit(1);
+
+      if (measured) {
+        await this.updateSyncCursor(tx, params, cursor.generation);
+        return { kind: 'skipped', reason: 'measured_session_present' };
+      }
+
+      if (params.bookFileId === null) return { kind: 'skipped', reason: 'book_file_not_found' };
+
+      const sessionId = params.buildSessionId(params.bookFileId, cursor.generation, params.counter);
+
+      const [fileRow] = await tx
+        .select({ bookId: books.id, libraryId: books.libraryId })
+        .from(bookFiles)
+        .innerJoin(books, eq(books.id, bookFiles.bookId))
+        .where(and(eq(bookFiles.id, params.bookFileId), eq(books.id, params.bookId)))
+        .limit(1);
+      if (!fileRow) return { kind: 'skipped', reason: 'book_file_not_found' };
+
+      const inserted = await tx
+        .insert(readingSessions)
+        .values({
+          userId: params.userId,
+          bookFileId: params.bookFileId,
+          bookId: params.bookId,
+          attemptId: sql`(select id from reading_attempts where user_id = ${params.userId} and book_id = ${params.bookId} and outcome is null and deleted_at is null limit 1)`,
+          sessionId,
+          source: params.source,
+          sourceDeviceKey: params.sourceDeviceKey,
+          startedAt,
+          endedAt: params.endedAt,
+          durationSeconds,
+          progressDelta: params.progressDelta,
+          endProgress: params.endProgress,
+        })
+        .onConflictDoNothing({ target: [readingSessions.userId, readingSessions.sessionId] })
+        .returning({ id: readingSessions.id });
+
+      await this.updateSyncCursor(tx, params, cursor.generation);
+      if (inserted.length === 0) return { kind: 'skipped', reason: 'duplicate_session_id' };
+
+      await this.upsertDailyStats(tx, {
+        userId: params.userId,
+        libraryId: fileRow.libraryId,
+        startedAt,
+        endedAt: params.endedAt,
+        durationSeconds,
+        progressDelta: params.progressDelta,
+        timeZone: params.timeZone,
+      });
+
+      return { kind: 'saved', durationSeconds, sessionId };
     });
   }
 
@@ -540,6 +719,101 @@ export class ReadingSessionRepository {
     await this.insertDailyStatsSegments(tx, userId, libraryId, segments, 'replace');
   }
 
+  private async updateSyncCursor(tx: Tx, params: RecordCumulativeReadingSessionParams, generation: number): Promise<void> {
+    await tx
+      .update(readingSessionSyncCursors)
+      .set({ counter: params.counter, generation, lastModified: params.endedAt, updatedAt: new Date() })
+      .where(
+        and(
+          eq(readingSessionSyncCursors.userId, params.userId),
+          eq(readingSessionSyncCursors.bookId, params.bookId),
+          eq(readingSessionSyncCursors.source, params.cursorSource),
+          eq(readingSessionSyncCursors.sourceDeviceKey, params.sourceDeviceKey),
+        ),
+      );
+  }
+
+  private async lockSyncSource(tx: Tx, userId: number, bookId: number, sourceDeviceKey: string): Promise<void> {
+    await tx.execute(sql`select pg_advisory_xact_lock(${userId}::int, hashtext(${`${sourceDeviceKey}:${bookId}`})::int)`);
+  }
+
+  private async deleteOverlappingSyncEstimates(
+    tx: Tx,
+    params: {
+      userId: number;
+      bookId: number;
+      libraryId: number;
+      source: ReadingSessionSource;
+      sourceDeviceKey: string;
+      estimateSessionIdPrefix: string;
+      startedAt: Date;
+      endedAt: Date;
+      timeZone: string;
+    },
+  ): Promise<void> {
+    const affectedDays = new Set<string>();
+    let cursor = 0;
+
+    for (;;) {
+      const rows = await tx
+        .select({
+          id: readingSessions.id,
+          startedAt: readingSessions.startedAt,
+          endedAt: readingSessions.endedAt,
+          durationSeconds: readingSessions.durationSeconds,
+          progressDelta: readingSessions.progressDelta,
+        })
+        .from(readingSessions)
+        .where(
+          and(
+            eq(readingSessions.userId, params.userId),
+            eq(readingSessions.bookId, params.bookId),
+            eq(readingSessions.source, params.source),
+            eq(readingSessions.sourceDeviceKey, params.sourceDeviceKey),
+            like(readingSessions.sessionId, `${params.estimateSessionIdPrefix}%`),
+            lt(readingSessions.startedAt, params.endedAt),
+            gt(readingSessions.endedAt, params.startedAt),
+            gt(readingSessions.id, cursor),
+          ),
+        )
+        .orderBy(readingSessions.id)
+        .limit(ESTIMATE_CLEANUP_PAGE_SIZE);
+
+      if (rows.length === 0) break;
+      cursor = rows[rows.length - 1]!.id;
+
+      for (const row of rows) {
+        for (const day of getReadingSessionDayKeys(
+          {
+            startedAt: row.startedAt,
+            endedAt: row.endedAt,
+            durationSeconds: row.durationSeconds,
+            progressDelta: row.progressDelta ?? null,
+          },
+          params.timeZone,
+        )) {
+          affectedDays.add(day);
+        }
+      }
+
+      await tx.delete(readingSessions).where(
+        and(
+          eq(readingSessions.userId, params.userId),
+          inArray(
+            readingSessions.id,
+            rows.map((row) => row.id),
+          ),
+        ),
+      );
+
+      if (rows.length < ESTIMATE_CLEANUP_PAGE_SIZE) break;
+    }
+
+    if (affectedDays.size > 0) {
+      await this.recomputeDailyStats(tx, params.userId, params.libraryId, [...affectedDays], params.timeZone);
+    }
+  }
+
   private async lockDailyStats(tx: Tx, userId: number, libraryId: number): Promise<void> {
     await tx.execute(sql`select pg_advisory_xact_lock(${userId}::int, ${libraryId}::int)`);
   }

server/src/modules/reading-session/reading-session.service.ts

@@ -11,7 +11,13 @@ import { AchievementEventsService, ACHIEVEMENT_EVENT_READING_SESSION_SAVED } fro
 import type { CreateManualReadingSessionDto } from './dto/create-manual-reading-session.dto';
 import type { ListBookReadingSessionsDto } from './dto/list-book-reading-sessions.dto';
 import type { SaveReadingSessionDto } from './dto/save-reading-session.dto';
-import { ReadingSessionRepository, type SaveReadingSessionResult } from './reading-session.repository';
+import {
+  ReadingSessionRepository,
+  type ReadingSessionSyncOptions,
+  type RecordCumulativeReadingSessionParams,
+  type RecordCumulativeReadingSessionResult,
+  type SaveReadingSessionResult,
+} from './reading-session.repository';
 
 @Injectable()
 export class ReadingSessionService {
@@ -27,7 +33,13 @@ export class ReadingSessionService {
     return resolveTimeZone((user.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC');
   }
 
-  async save(fileId: number, dto: SaveReadingSessionDto, user: RequestUser, source: ReadingSessionSource = 'web'): Promise<void> {
+  async save(
+    fileId: number,
+    dto: SaveReadingSessionDto,
+    user: RequestUser,
+    source: ReadingSessionSource = 'web',
+    sync?: ReadingSessionSyncOptions,
+  ): Promise<void> {
     const event = 'reading_session.save';
     const startedAtMs = Date.now();
     this.logger.log(
@@ -52,7 +64,7 @@ export class ReadingSessionService {
       // span to prevent the client from reporting more time than physically elapsed.
       const durationSeconds = Math.min(dto.durationSeconds, wallClockSeconds);
 
-      const result = await this.repo.saveSession(
+      const saveArgs = [
         user.id,
         fileId,
         dto.sessionId,
@@ -63,7 +75,8 @@ export class ReadingSessionService {
         dto.endProgress ?? null,
         source,
         this.resolveUserTimeZone(user),
-      );
+      ] as const;
+      const result = sync ? await this.repo.saveSession(...saveArgs, sync) : await this.repo.saveSession(...saveArgs);
 
       this.logger.log(
         `[${event}] [end] fileId=${fileId} userId=${user.id} sessionId=${dto.sessionId} durationMs=${Date.now() - startedAtMs} outcome=${result.kind}${result.kind === 'skipped' ? ` reason=${result.reason}` : ''} - reading session save completed`,
@@ -148,6 +161,26 @@ export class ReadingSessionService {
     return result;
   }
 
+  async recordCumulativeSyncedSession(params: RecordCumulativeReadingSessionParams): Promise<RecordCumulativeReadingSessionResult> {
+    const result = await this.repo.recordCumulativeSyncedSession(params);
+
+    if (result.kind === 'saved' && params.bookFileId !== null) {
+      const startedAt = new Date(params.endedAt.getTime() - result.durationSeconds * 1000);
+      this.achievementEvents.emit(ACHIEVEMENT_EVENT_READING_SESSION_SAVED, {
+        userId: params.userId,
+        bookFileId: params.bookFileId,
+        durationSeconds: result.durationSeconds,
+        startedAt,
+        endedAt: params.endedAt,
+        progressDelta: params.progressDelta,
+        endProgress: params.endProgress,
+        timezone: params.timeZone,
+      });
+    }
+
+    return result;
+  }
+
   /**
    * Retires the estimates a device made for reading the same device has since reported real
    * page timings for. Prefixes are supplied by the caller, which owns both id formats.