From 24882ef269eff34038c1777045d0faae82b1d0bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?V=C3=A1clav=20Brzezina?= Date: Tue, 13 Oct 2026 15:07:23 +0200 Subject: [PATCH] fix(ingestion): keep synthetic activity timestamps out of the sync cursor --- .../modules/ingestion/ingestion.service.ts | 27 +++++----- .../src/modules/ingestion/sync-cursor.spec.ts | 36 +++++++++++++ .../src/modules/ingestion/sync-cursor.ts | 20 ++++++++ packages/core/src/types.ts | 7 +++ .../provider-arr/src/arr.provider.spec.ts | 50 +++++++++++++++++++ packages/provider-arr/src/arr.provider.ts | 2 + 6 files changed, 130 insertions(+), 12 deletions(-) create mode 100644 apps/backend/src/modules/ingestion/sync-cursor.spec.ts create mode 100644 apps/backend/src/modules/ingestion/sync-cursor.ts create mode 100644 packages/provider-arr/src/arr.provider.spec.ts diff --git a/apps/backend/src/modules/ingestion/ingestion.service.ts b/apps/backend/src/modules/ingestion/ingestion.service.ts index 66dd43d..a2c5960 100644 --- a/apps/backend/src/modules/ingestion/ingestion.service.ts +++ b/apps/backend/src/modules/ingestion/ingestion.service.ts @@ -19,6 +19,8 @@ import { IssuesService } from '../issues/issues.service'; import { LogsGateway } from '../logs/logs.gateway'; import { SessionsGateway } from '../sessions/sessions.gateway'; +import { computeSyncCursor } from './sync-cursor'; + import type { MediaServerProvider, NormalizedSession, NormalizedActivity } from '@logarr/core'; import type { EmbySession, @@ -241,18 +243,19 @@ export class IngestionService implements OnModuleInit, OnModuleDestroy { // Update last sync timestamp in database (persistence!) if (insertedCount > 0) { - const maxTimestamp = activities.reduce( - (max, a) => (a.timestamp > max ? a.timestamp : max), - activities[0]!.timestamp - ); - - await this.db - .update(schema.servers) - .set({ - lastActivitySync: maxTimestamp, - updatedAt: new Date(), - }) - .where(eq(schema.servers.id, server.id)); + // Synthetic rows (detection-time stamps) must not move the cursor; + // if nothing qualifies, leave it unchanged. + const nextCursor = computeSyncCursor(activities); + + if (nextCursor) { + await this.db + .update(schema.servers) + .set({ + lastActivitySync: nextCursor, + updatedAt: new Date(), + }) + .where(eq(schema.servers.id, server.id)); + } } } diff --git a/apps/backend/src/modules/ingestion/sync-cursor.spec.ts b/apps/backend/src/modules/ingestion/sync-cursor.spec.ts new file mode 100644 index 0000000..2191cb1 --- /dev/null +++ b/apps/backend/src/modules/ingestion/sync-cursor.spec.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; + +import { computeSyncCursor } from './sync-cursor'; + +import type { NormalizedActivity } from '@logarr/core'; + +const act = (iso: string, advancesSyncCursor?: boolean): NormalizedActivity => ({ + id: iso, + type: 't', + name: 'n', + severity: 'info', + timestamp: new Date(iso), + ...(advancesSyncCursor === undefined ? {} : { advancesSyncCursor }), +}); + +describe('computeSyncCursor', () => { + it('uses the max timestamp of event-time rows, ignoring synthetic rows', () => { + const result = computeSyncCursor([ + act('2026-01-01T00:00:00Z'), + act('2026-01-03T00:00:00Z', true), + act('2026-01-02T00:00:00Z'), + act('2026-06-01T00:00:00Z', false), + ]); + expect(result).toEqual(new Date('2026-01-03T00:00:00Z')); + }); + + it('returns undefined when all rows are synthetic', () => { + expect( + computeSyncCursor([act('2026-06-01T00:00:00Z', false), act('2026-06-02T00:00:00Z', false)]) + ).toBeUndefined(); + }); + + it('returns undefined for an empty batch', () => { + expect(computeSyncCursor([])).toBeUndefined(); + }); +}); diff --git a/apps/backend/src/modules/ingestion/sync-cursor.ts b/apps/backend/src/modules/ingestion/sync-cursor.ts new file mode 100644 index 0000000..0352d87 --- /dev/null +++ b/apps/backend/src/modules/ingestion/sync-cursor.ts @@ -0,0 +1,20 @@ +import type { NormalizedActivity } from '@logarr/core'; + +/** + * Compute the next `lastActivitySync` value from a batch of activities. + * + * Only activities whose timestamp is an event time count (`advancesSyncCursor` + * !== false). Synthetic rows stamped with the detection time are ignored so they + * can't move the cursor to "now" and hide history not returned in this poll. + * + * @returns the max qualifying timestamp, or undefined if none qualify + * (caller should leave the stored cursor unchanged). + */ +export function computeSyncCursor(activities: readonly NormalizedActivity[]): Date | undefined { + let max: Date | undefined; + for (const activity of activities) { + if (activity.advancesSyncCursor === false) continue; + if (!max || activity.timestamp > max) max = activity.timestamp; + } + return max; +} diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index a55b7e3..1e1dd04 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -239,6 +239,13 @@ export interface NormalizedActivity { readonly itemId?: string | undefined; readonly timestamp: Date; readonly metadata?: Record | undefined; + /** + * Whether this activity's timestamp may move the provider's `since` sync cursor. + * Defaults to true. Set to false for synthetic rows whose timestamp is the + * detection time rather than the event time (health checks, connection status, + * queue warnings), so they don't skip history that wasn't returned this poll. + */ + readonly advancesSyncCursor?: boolean | undefined; } // ============================================================================= diff --git a/packages/provider-arr/src/arr.provider.spec.ts b/packages/provider-arr/src/arr.provider.spec.ts new file mode 100644 index 0000000..09a4aba --- /dev/null +++ b/packages/provider-arr/src/arr.provider.spec.ts @@ -0,0 +1,50 @@ +import { describe, it, expect } from 'vitest'; + +import { ArrBaseProvider } from './arr.provider.js'; + +import type { ArrHealthCheck, ArrHistoryRecordBase, ArrQueueItemBase } from './arr.types.js'; +import type { NormalizedActivity } from '@logarr/core'; + +class TestArrProvider extends ArrBaseProvider { + readonly id = 'test'; + readonly name = 'Test'; + protected getHistoryRecords(): Promise { + return Promise.resolve([]); + } + protected normalizeHistoryRecord(): NormalizedActivity { + throw new Error('not used'); + } + protected getMediaType(): 'series' { + return 'series'; + } + health(checks: readonly ArrHealthCheck[]): NormalizedActivity[] { + return this.normalizeHealthChecks(checks); + } + queue(items: readonly ArrQueueItemBase[]): NormalizedActivity[] { + return this.normalizeQueueItems(items); + } +} + +describe('ArrBaseProvider synthetic activities', () => { + it('flags health activities as not advancing the sync cursor', () => { + const result = new TestArrProvider().health([ + { source: 'IndexerCheck', type: 'warning', message: 'm', wikiUrl: '' }, + ] as unknown as ArrHealthCheck[]); + expect(result).toHaveLength(1); + expect(result[0]!.advancesSyncCursor).toBe(false); + }); + + it('flags queue activities as not advancing the sync cursor', () => { + const result = new TestArrProvider().queue([ + { + title: 'x', + downloadId: 'd1', + trackedDownloadStatus: 'warning', + errorMessage: 'bad', + statusMessages: [], + }, + ] as unknown as ArrQueueItemBase[]); + expect(result).toHaveLength(1); + expect(result[0]!.advancesSyncCursor).toBe(false); + }); +}); diff --git a/packages/provider-arr/src/arr.provider.ts b/packages/provider-arr/src/arr.provider.ts index 20b3961..8d405c1 100644 --- a/packages/provider-arr/src/arr.provider.ts +++ b/packages/provider-arr/src/arr.provider.ts @@ -278,6 +278,7 @@ export abstract class ArrBaseProvider implements MediaServerProvider { overview: check.message, severity: check.type === 'error' ? 'error' : ('warn' as const), timestamp: new Date(), + advancesSyncCursor: false, })); } @@ -295,6 +296,7 @@ export abstract class ArrBaseProvider implements MediaServerProvider { item.errorMessage ?? item.statusMessages.map((m) => m.messages.join(', ')).join('; '), severity: 'warn' as const, timestamp: new Date(), + advancesSyncCursor: false, })); }