From 20807d2d3d6b25d6df2e4b5b502c18bca784180a Mon Sep 17 00:00:00 2001 From: Dmitrii Date: Mon, 28 Sep 2026 13:51:25 +0000 Subject: [PATCH] fix(core): Enforce one instance report per day with a unique index (#39536) Co-authored-by: konstantintieber --- docs/generated/postgres-schema/README.md | 3 +- .../public.instance_monitoring_report.md | 3 + docs/generated/sqlite-schema/README.md | 3 +- .../instance_monitoring_report.md | 7 +- ...AddReportDateToInstanceMonitoringReport.ts | 76 ++++++ .../LOCAL_VERIFICATION.md | 4 +- .../modules/instance-reporting.ee/RETRIES.md | 2 +- ...stance-reporting-scheduler.service.test.ts | 12 +- .../instance-reporting.integration.test.ts | 69 +++-- .../instance-reporting.service.test.ts | 64 ++--- .../entities/instance-monitoring-report.ts | 9 +- ...ring-report.repository.integration.test.ts | 116 ++++++-- .../instance-monitoring-report.repository.ts | 45 ++-- .../instance-reporting-scheduler.service.ts | 39 ++- .../instance-reporting.service.ts | 34 +-- ...date-to-instance-monitoring-report.test.ts | 253 ++++++++++++++++++ 16 files changed, 603 insertions(+), 136 deletions(-) create mode 100644 packages/@n8n/db/src/migrations/common/1790583238183-AddReportDateToInstanceMonitoringReport.ts create mode 100644 packages/cli/test/migration/1790583238183-add-report-date-to-instance-monitoring-report.test.ts diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index f55dc61540e..800f9b4363a 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -90,7 +90,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.instance_ai_threads](public.instance_ai_threads.md) | 7 | | BASE TABLE | | [public.instance_ai_workflow_snapshots](public.instance_ai_workflow_snapshots.md) | 7 | | BASE TABLE | | [public.instance_credential_assignment](public.instance_credential_assignment.md) | 4 | | BASE TABLE | -| [public.instance_monitoring_report](public.instance_monitoring_report.md) | 9 | | BASE TABLE | +| [public.instance_monitoring_report](public.instance_monitoring_report.md) | 10 | | BASE TABLE | | [public.instance_version_history](public.instance_version_history.md) | 5 | | BASE TABLE | | [public.invalid_auth_token](public.invalid_auth_token.md) | 2 | | BASE TABLE | | [public.mcp_registry_server](public.mcp_registry_server.md) | 7 | | BASE TABLE | @@ -1246,6 +1246,7 @@ erDiagram uuid id timestamp_3__with_time_zone lastAttemptAt text lastError + varchar_10_ reportDate varchar_64_ status timestamp_3__with_time_zone updatedAt } diff --git a/docs/generated/postgres-schema/public.instance_monitoring_report.md b/docs/generated/postgres-schema/public.instance_monitoring_report.md index 670d4a1a5e8..f2348148da3 100644 --- a/docs/generated/postgres-schema/public.instance_monitoring_report.md +++ b/docs/generated/postgres-schema/public.instance_monitoring_report.md @@ -11,6 +11,7 @@ | id | uuid | | false | | | UUID; travels with the payload as its `batchId`. | | lastAttemptAt | timestamp(3) with time zone | | true | | | When the last attempt finished; NULL before the first. Paces retries across a restart. | | lastError | text | | true | | | Message of the most recent delivery failure. | +| reportDate | varchar(10) | | true | | | UTC day the report was created on, as YYYY-MM-DD. Unique, so one report is created a day. Not the days the report covers. NULL only on a legacy row that shared its day with an earlier one. | | status | varchar(64) | 'pending'::character varying | false | | | Skipped means the instance stopped trying that day, not that the numbers were lost: only a delivered report crosses a day off, so a skipped day is covered by the next report. | | updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | @@ -31,6 +32,7 @@ | Name | Definition | | ---- | ---------- | +| IDX_instance_monitoring_report_reportDate | CREATE UNIQUE INDEX "IDX_instance_monitoring_report_reportDate" ON public.instance_monitoring_report USING btree ("reportDate") WHERE ("reportDate" IS NOT NULL) | | PK_4a5cb8aa51c67e4f5a7eb0f9509 | CREATE UNIQUE INDEX "PK_4a5cb8aa51c67e4f5a7eb0f9509" ON public.instance_monitoring_report USING btree (id) | ## Relations @@ -47,6 +49,7 @@ erDiagram uuid id timestamp_3__with_time_zone lastAttemptAt text lastError + varchar_10_ reportDate varchar_64_ status timestamp_3__with_time_zone updatedAt } diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 90d3fc246a8..e78996afc49 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -90,7 +90,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [instance_ai_threads](instance_ai_threads.md) | 7 | | table | | [instance_ai_workflow_snapshots](instance_ai_workflow_snapshots.md) | 7 | | table | | [instance_credential_assignment](instance_credential_assignment.md) | 4 | | table | -| [instance_monitoring_report](instance_monitoring_report.md) | 9 | | table | +| [instance_monitoring_report](instance_monitoring_report.md) | 10 | | table | | [instance_version_history](instance_version_history.md) | 5 | | table | | [invalid_auth_token](invalid_auth_token.md) | 2 | | table | | [mcp_registry_server](mcp_registry_server.md) | 7 | | table | @@ -1232,6 +1232,7 @@ erDiagram varchar id PK datetime_3_ lastAttemptAt TEXT lastError + varchar_10_ reportDate varchar_64_ status datetime_3_ updatedAt } diff --git a/docs/generated/sqlite-schema/instance_monitoring_report.md b/docs/generated/sqlite-schema/instance_monitoring_report.md index b73a8fe4546..a7d4ab9ba64 100644 --- a/docs/generated/sqlite-schema/instance_monitoring_report.md +++ b/docs/generated/sqlite-schema/instance_monitoring_report.md @@ -6,7 +6,7 @@ Table Definition ```sql -CREATE TABLE "instance_monitoring_report" ("id" varchar PRIMARY KEY NOT NULL, "dataPoints" text NOT NULL, "status" varchar(64) NOT NULL DEFAULT ('pending'), "deliveredAt" datetime(3), "attempts" integer NOT NULL DEFAULT (0), "lastAttemptAt" datetime(3), "lastError" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_instance_monitoring_report_status" CHECK ("status" IN ('pending', 'delivered', 'skipped_after_max_retries'))) +CREATE TABLE "instance_monitoring_report" ("id" varchar PRIMARY KEY NOT NULL, "dataPoints" text NOT NULL, "status" varchar(64) NOT NULL DEFAULT ('pending'), "deliveredAt" datetime(3), "attempts" integer NOT NULL DEFAULT (0), "lastAttemptAt" datetime(3), "lastError" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "reportDate" varchar(10), CONSTRAINT "CHK_instance_monitoring_report_status" CHECK (("status" IN ('pending', 'delivered', 'skipped_after_max_retries')))) ``` @@ -22,6 +22,7 @@ CREATE TABLE "instance_monitoring_report" ("id" varchar PRIMARY KEY NOT NULL, "d | id | varchar | | false | | | | | lastAttemptAt | datetime(3) | | true | | | | | lastError | TEXT | | true | | | | +| reportDate | varchar(10) | | true | | | | | status | varchar(64) | 'pending' | false | | | | | updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | @@ -29,7 +30,7 @@ CREATE TABLE "instance_monitoring_report" ("id" varchar PRIMARY KEY NOT NULL, "d | Name | Type | Definition | | ---- | ---- | ---------- | -| - | CHECK | CHECK ("status" IN ('pending', 'delivered', 'skipped_after_max_retries')) | +| - | CHECK | CHECK (("status" IN ('pending', 'delivered', 'skipped_after_max_retries'))) | | id | PRIMARY KEY | PRIMARY KEY (id) | | sqlite_autoindex_instance_monitoring_report_1 | PRIMARY KEY | PRIMARY KEY (id) | @@ -37,6 +38,7 @@ CREATE TABLE "instance_monitoring_report" ("id" varchar PRIMARY KEY NOT NULL, "d | Name | Definition | | ---- | ---------- | +| IDX_instance_monitoring_report_reportDate | CREATE UNIQUE INDEX "IDX_instance_monitoring_report_reportDate" ON "instance_monitoring_report" ("reportDate") WHERE "reportDate" IS NOT NULL | | sqlite_autoindex_instance_monitoring_report_1 | PRIMARY KEY (id) | ## Relations @@ -53,6 +55,7 @@ erDiagram varchar id PK datetime_3_ lastAttemptAt TEXT lastError + varchar_10_ reportDate varchar_64_ status datetime_3_ updatedAt } diff --git a/packages/@n8n/db/src/migrations/common/1790583238183-AddReportDateToInstanceMonitoringReport.ts b/packages/@n8n/db/src/migrations/common/1790583238183-AddReportDateToInstanceMonitoringReport.ts new file mode 100644 index 00000000000..3f47d9a1bfa --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1790583238183-AddReportDateToInstanceMonitoringReport.ts @@ -0,0 +1,76 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +const TABLE_NAME = 'instance_monitoring_report'; +const COLUMN_NAME = 'reportDate'; + +export class AddReportDateToInstanceMonitoringReport1790583238183 implements ReversibleMigration { + async up(context: MigrationContext) { + const { addColumns, column, createIndex } = context.schemaBuilder; + + await addColumns( + TABLE_NAME, + [ + column(COLUMN_NAME) + .varchar(10) + .comment( + 'UTC day the report was created on, as YYYY-MM-DD. Unique, so one report is created a day. Not the days the report covers. NULL only on a legacy row that shared its day with an earlier one.', + ), + ], + { recreatesOnSqlite: true }, + ); + + await this.backfillReportDate(context); + await this.clearDuplicateDays(context); + + await createIndex( + TABLE_NAME, + [COLUMN_NAME], + true, + undefined, + `${context.escape.columnName(COLUMN_NAME)} IS NOT NULL`, + ); + } + + async down({ schemaBuilder: { dropIndex, dropColumns } }: MigrationContext) { + await dropIndex(TABLE_NAME, [COLUMN_NAME]); + await dropColumns(TABLE_NAME, [COLUMN_NAME], { recreatesOnSqlite: true }); + } + + private async backfillReportDate({ escape, isPostgres, runQuery }: MigrationContext) { + const table = escape.tableName(TABLE_NAME); + const reportDate = escape.columnName(COLUMN_NAME); + const createdAt = escape.columnName('createdAt'); + + const utcDay = isPostgres + ? `to_char(${createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD')` + : `strftime('%Y-%m-%d', ${createdAt})`; + + await runQuery(`UPDATE ${table} SET ${reportDate} = ${utcDay}`); + } + + /** Keeps the day on its most settled row, so the unique index can be built. */ + private async clearDuplicateDays({ escape, runQuery }: MigrationContext) { + const table = escape.tableName(TABLE_NAME); + const id = escape.columnName('id'); + const reportDate = escape.columnName(COLUMN_NAME); + const status = escape.columnName('status'); + const createdAt = escape.columnName('createdAt'); + + await runQuery( + `UPDATE ${table} + SET ${reportDate} = NULL, + ${status} = CASE ${status} WHEN 'pending' THEN 'skipped_after_max_retries' ELSE ${status} END + WHERE ${id} IN ( + SELECT ${id} FROM ( + SELECT ${id}, ROW_NUMBER() OVER ( + PARTITION BY ${reportDate} + ORDER BY + CASE ${status} WHEN 'delivered' THEN 0 WHEN 'skipped_after_max_retries' THEN 1 ELSE 2 END, + ${createdAt} DESC, + ${id} DESC + ) AS day_rank FROM ${table} + ) ranked WHERE day_rank > 1 + )`, + ); + } +} diff --git a/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md b/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md index 8f36ee405e3..58160100308 100644 --- a/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md +++ b/packages/cli/src/modules/instance-reporting.ee/LOCAL_VERIFICATION.md @@ -165,7 +165,7 @@ yesterday's UTC date, not today's. ```bash sqlite3 -header "$N8N_DB" \ - "SELECT id, createdAt, deliveredAt, attempts, lastError, dataPoints + "SELECT id, reportDate, createdAt, deliveredAt, attempts, lastError, dataPoints FROM instance_monitoring_report ORDER BY createdAt DESC LIMIT 5;" ``` @@ -179,7 +179,7 @@ Run these after step 4. Each is short. | # | Scenario | Steps | Expected | |---|---|---|---| -| 5.1 | No second report the same day | Restart the instance | No new request, no new row. `hasDeliveredToday` short-circuits the tick | +| 5.1 | No second report the same day | Restart the instance | No new request, no new row. `hasSettledToday` short-circuits the tick | | 5.2 | Retry resends the same measurement | Make the receiver answer 500. Delete today's delivered row, then restart | Request arrives, `lastError` holds `rejected with status 500`, `deliveredAt` NULL, `attempts` grows. Retries land ~5 minutes apart, 3 attempts in total, then `Giving up on the instance report for today`. Every retry carries the **same** `batchId` and the same values — no re-measurement | | 5.3 | Recovery keeps the pending row | During 5.2, switch the receiver back to 201 before the third attempt | The next attempt reuses the pending row and marks it delivered. No second row for the day | | 5.4 | Untrusted certificate | Set `N8N_LICENSE_CERT` to a certificate the receiver does not trust (any well-formed one from another CA), clear today's row, restart | The receiver answers 401, delivery fails, `lastError` names the status. Nothing is marked delivered | diff --git a/packages/cli/src/modules/instance-reporting.ee/RETRIES.md b/packages/cli/src/modules/instance-reporting.ee/RETRIES.md index 6fc85e5d009..f405e27a48a 100644 --- a/packages/cli/src/modules/instance-reporting.ee/RETRIES.md +++ b/packages/cli/src/modules/instance-reporting.ee/RETRIES.md @@ -51,7 +51,7 @@ next slot has passed, the retry stops: the row is marked - The row is created once, as `pending`, with its data points already measured. A retry changes only `attempts`, `lastAttemptAt`, `lastError` and, in the end, `status`. The data points and the `batchId` stay the same. -- `InstanceReportingService.msUntilRetryAllowed()` reads `lastAttemptAt` from +- `InstanceReportingScheduler.msUntilRetryAllowed()` reads `lastAttemptAt` from the row. The scheduler waits `RETRY_DELAY_MS` (5 minutes) between attempts. - The budget is `MAX_ATTEMPTS` (3). The attempt that spends the last one flips the row to `skipped_after_max_retries` at once. There is no fourth attempt, diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts index 2572890f56b..77d798e45fa 100644 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting-scheduler.service.test.ts @@ -6,6 +6,7 @@ import { mock } from 'vitest-mock-extended'; import type { EventService } from '@/events/event.service'; +import type { InstanceMonitoringReport } from '../database/entities/instance-monitoring-report'; import type { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository'; import { InstanceReportingScheduler } from '../instance-reporting-scheduler.service'; import type { InstanceReportingSettingsService } from '../instance-reporting-settings.service'; @@ -38,8 +39,6 @@ function makeHarness({ autoStart = true, } = {}): Harness { const reportingService = mock(); - // No attempt has been made yet, so nothing is holding the next one back. - reportingService.msUntilRetryAllowed.mockResolvedValue(0); const reportRepository = mock(); reportRepository.hasSettledToday.mockResolvedValue(false); @@ -285,8 +284,13 @@ describe('InstanceReportingScheduler', () => { // A crash between attempts would otherwise let the fresh process attempt at // once, and a crash loop would spend the whole budget in seconds. vi.setSystemTime(new Date(AFTER_SLOT)); - const { scheduler, reportingService } = makeHarness(); - reportingService.msUntilRetryAllowed.mockResolvedValueOnce(2 * Time.minutes.toMilliseconds); + const { scheduler, reportingService, reportRepository } = makeHarness(); + reportRepository.findPending.mockResolvedValue( + mock({ + createdAt: new Date(AFTER_SLOT), + lastAttemptAt: new Date(Date.now() - 3 * Time.minutes.toMilliseconds), + }), + ); scheduler.init(); await settle(); diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts index 552d7bc6be9..eddc894cf71 100644 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.integration.test.ts @@ -21,7 +21,10 @@ import { InsightsService } from '@/modules/insights/insights.service'; import { OwnershipService } from '@/services/ownership.service'; import { createOwner } from '@test-integration/db/users'; -import type { InstanceReportDataPoint } from '../database/entities/instance-monitoring-report'; +import type { + InstanceMonitoringReport, + InstanceReportDataPoint, +} from '../database/entities/instance-monitoring-report'; import { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository'; import { InstanceReportingScheduler } from '../instance-reporting-scheduler.service'; import { InstanceReportingSettingsService } from '../instance-reporting-settings.service'; @@ -190,25 +193,25 @@ describe('instance reporting retries', () => { ); } - /** - * The database stamps `createdAt` with the real clock, not the fake one, so - * a row from a fake "yesterday" reads as today's. Pin it to the day meant. - */ - async function stampCreatedAt(id: string, createdAt: Date) { - await repository.update({ id }, { createdAt }); + async function createPendingOn(createdAt: Date, dataPoints: InstanceReportDataPoint[]) { + const report = await repository.createPending(dataPoints, createdAt); + if (!report) throw new Error(`A report was already created on ${createdAt.toISOString()}`); + // The database stamps `createdAt` with the real clock, not the fake one. + await repository.update({ id: report.id }, { createdAt }); + + return report; } /** A delivered report that covers `day`, generated the morning after it. */ async function seedDeliveredReport(day: string) { - const report = await repository.createPending([ - { kind: 'cumulative', name: 'billableExecutions', value: 0 }, - { kind: 'daily', name: 'billableExecutions', value: 3, date: day }, - ]); - await repository.markDelivered(report.id, new Date()); - await stampCreatedAt( - report.id, + const report = await createPendingOn( new Date(new Date(`${day}T07:43:00.000Z`).getTime() + Time.days.toMilliseconds), + [ + { kind: 'cumulative', name: 'billableExecutions', value: 0 }, + { kind: 'daily', name: 'billableExecutions', value: 3, date: day }, + ], ); + await repository.markDelivered(report.id, new Date()); return report; } @@ -232,7 +235,7 @@ describe('instance reporting retries', () => { createdAt, }: { attempts: number; lastAttemptAt: Date; createdAt: Date }, ) { - const report = await repository.createPending([ + const report = await createPendingOn(createdAt, [ { kind: 'cumulative', name: 'billableExecutions', value: 0 }, { kind: 'daily', name: 'billableExecutions', value: 5, date: day }, ]); @@ -240,7 +243,6 @@ describe('instance reporting retries', () => { { id: report.id }, { attempts, lastError: 'ECONNREFUSED', lastAttemptAt }, ); - await stampCreatedAt(report.id, createdAt); return report; } @@ -328,8 +330,6 @@ describe('instance reporting retries', () => { const [abandoned] = await repository.find({ where: { status: 'skipped_after_max_retries' } }); expect(abandoned).toMatchObject({ attempts: MAX_ATTEMPTS }); expect(dailyPoints(sentPayload(harness, 0))).toEqual([{ date: '2026-03-25', value: 5 }]); - // Keep the abandoned row on the day it was made, or tomorrow reads as settled too. - await stampCreatedAt(abandoned.id, new Date(AFTER_SLOT)); // The next pass finds the day settled and moves on to tomorrow's slot. await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS); @@ -369,8 +369,6 @@ describe('instance reporting retries', () => { const [rejected] = await repository.find({ where: { status: 'skipped_after_max_retries' } }); expect(rejected).toMatchObject({ attempts: 1 }); expectArmedFor(harness, NEXT_SLOT); - // Keep the rejected row on the day it was made, or tomorrow reads as settled too. - await stampCreatedAt(rejected.id, new Date(AFTER_SLOT)); await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS); expect(harness.httpRequest).toHaveBeenCalledTimes(1); @@ -700,4 +698,35 @@ describe('instance reporting retries', () => { const dates = await deliveredDailyDates(); expect(new Set(dates).size).toBe(dates.length); }); + + test('retries after the delay and sends the report of a process that created it first and stopped', async () => { + const otherPoints: InstanceReportDataPoint[] = [ + { kind: 'cumulative', name: 'billableExecutions', value: 999 }, + ]; + let other: InstanceMonitoringReport | undefined; + + // Another process inserts today's report after this one found nothing pending. + const insights = Container.get(InsightsService); + const measure = insights.getInsightsByTime.bind(insights); + vi.spyOn(insights, 'getInsightsByTime').mockImplementation(async (args) => { + other = await createPendingOn(new Date(), otherPoints); + return await measure(args); + }); + + const harness = makeHarness([accepted()]); + harness.scheduler.start(); + await armed(harness, 1); + + expect(harness.httpRequest).not.toHaveBeenCalled(); + expect(harness.scheduleNext).toHaveBeenLastCalledWith(RETRY_DELAY_MS); + + await vi.advanceTimersByTimeAsync(RETRY_DELAY_MS); + await armed(harness, 2); + + expect(harness.httpRequest).toHaveBeenCalledTimes(1); + expect(sentPayload(harness, 0)).toMatchObject({ batchId: other?.id, dataPoints: otherPoints }); + await expect(repository.find()).resolves.toEqual([ + expect.objectContaining({ id: other?.id, status: 'delivered' }), + ]); + }); }); diff --git a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts index e18c0b3c60f..f900357e43b 100644 --- a/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/__tests__/instance-reporting.service.test.ts @@ -19,7 +19,10 @@ import type { OwnershipService } from '@/services/ownership.service'; import type { InstanceMonitoringReport } from '../database/entities/instance-monitoring-report'; import type { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository'; import { InstanceReportingConfig } from '../instance-reporting.config'; -import { InstanceReportingService } from '../instance-reporting.service'; +import { + InstanceReportAlreadyCreatedError, + InstanceReportingService, +} from '../instance-reporting.service'; vi.mock('@/constants', async (importOriginal) => ({ ...(await importOriginal()), @@ -330,13 +333,27 @@ describe('InstanceReportingService', () => { await service.sendReport(); - expect(reportRepository.createPending).toHaveBeenCalledWith([ - { kind: 'cumulative', name: 'billableExecutions', value: 815 }, - { kind: 'daily', name: 'billableExecutions', value: 42, date: REPORT_DATE }, - ]); + expect(reportRepository.createPending).toHaveBeenCalledWith( + [ + { kind: 'cumulative', name: 'billableExecutions', value: 815 }, + { kind: 'daily', name: 'billableExecutions', value: 42, date: REPORT_DATE }, + ], + new Date('2026-03-26T07:42:00.000Z'), + ); expect(reportRepository.markDelivered).toHaveBeenCalledWith(BATCH_ID, expect.any(Date)); }); + test('throws without sending when another process already created the report for today', async () => { + const { service, reportRepository, http } = makeHarness(); + reportRepository.createPending.mockResolvedValue(null); + + await expect(service.sendReport()).rejects.toThrow(InstanceReportAlreadyCreatedError); + + expect(http.request).not.toHaveBeenCalled(); + expect(reportRepository.recordFailure).not.toHaveBeenCalled(); + expect(reportRepository.markDelivered).not.toHaveBeenCalled(); + }); + test('records the failure and rethrows when the request fails', async () => { const { service, reportRepository, http } = makeHarness(); vi.mocked(http.request).mockRejectedValue(new Error('Network error')); @@ -515,43 +532,6 @@ describe('InstanceReportingService', () => { }); }); - describe('msUntilRetryAllowed', () => { - const now = new Date('2026-03-26T07:42:00.000Z'); - - test('allows an attempt when no report is pending', async () => { - const { service, reportRepository } = makeHarness(); - reportRepository.findPending.mockResolvedValue(null); - - await expect(service.msUntilRetryAllowed(now)).resolves.toBe(0); - }); - - test('allows the first attempt, which has nothing to wait for', async () => { - const { service, reportRepository } = makeHarness(); - reportRepository.findPending.mockResolvedValue(makeReport({ lastAttemptAt: null })); - - await expect(service.msUntilRetryAllowed(now)).resolves.toBe(0); - }); - - test('returns the remaining wait when the last attempt was recent', async () => { - const { service, reportRepository } = makeHarness(); - reportRepository.findPending.mockResolvedValue( - makeReport({ lastAttemptAt: new Date('2026-03-26T07:40:00.000Z') }), - ); - - // Two of the five minutes are spent, so three remain. - await expect(service.msUntilRetryAllowed(now)).resolves.toBe(3 * 60 * 1000); - }); - - test('allows an attempt once the wait has passed', async () => { - const { service, reportRepository } = makeHarness(); - reportRepository.findPending.mockResolvedValue( - makeReport({ lastAttemptAt: new Date('2026-03-26T07:30:00.000Z') }), - ); - - await expect(service.msUntilRetryAllowed(now)).resolves.toBe(0); - }); - }); - describe('missed days', () => { beforeEach(() => { vi.useFakeTimers(); diff --git a/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts b/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts index adf8b424d0d..13dfe68985a 100644 --- a/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts +++ b/packages/cli/src/modules/instance-reporting.ee/database/entities/instance-monitoring-report.ts @@ -1,5 +1,5 @@ import { DateTimeColumn, JsonColumn, WithTimestamps } from '@n8n/db'; -import { Column, Entity, PrimaryColumn } from '@n8n/typeorm'; +import { Column, Entity, Index, PrimaryColumn } from '@n8n/typeorm'; /** One measurement in a report; the shape the receiver accepts. */ export type InstanceReportDataPoint = @@ -29,12 +29,19 @@ export type InstanceReportStatus = 'pending' | 'delivered' | 'skipped_after_max_ * The retry state — {@link status}, {@link attempts} and {@link lastAttemptAt} — * lives here rather than in the scheduler's memory, so a restart resumes the * report's budget instead of granting a fresh one. + * + * At most one row is created per UTC day; see {@link reportDate}. */ @Entity() +@Index(['reportDate'], { unique: true, where: '"reportDate" IS NOT NULL' }) export class InstanceMonitoringReport extends WithTimestamps { @PrimaryColumn('uuid') id: string; + /** UTC day the report was created on, as `YYYY-MM-DD`; not a day it covers. */ + @Column({ type: 'varchar', length: 10, nullable: true }) + reportDate: string | null; + /** The data point array exactly as sent. */ @JsonColumn() dataPoints: InstanceReportDataPoint[]; diff --git a/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts b/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts index a6ee5457450..43be3fcc102 100644 --- a/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts +++ b/packages/cli/src/modules/instance-reporting.ee/database/repositories/__tests__/instance-monitoring-report.repository.integration.test.ts @@ -28,9 +28,16 @@ describe('InstanceMonitoringReportRepository', () => { await testDb.terminate(); }); + async function createOn(date: string | Date, dataPoints = DATA_POINTS) { + const report = await repository.createPending(dataPoints, new Date(date)); + if (!report) throw new Error(`A report was already created on the day of ${String(date)}`); + + return report; + } + describe('createPending', () => { test('records the measurement with a generated id, undelivered', async () => { - const report = await repository.createPending(DATA_POINTS); + const report = await createOn(new Date()); expect(report.id).toEqual(expect.any(String)); expect(report.deliveredAt).toBeNull(); @@ -40,11 +47,39 @@ describe('InstanceMonitoringReportRepository', () => { dataPoints: DATA_POINTS, }); }); + + test.each([ + ['2026-03-25T00:00:00.000Z', '2026-03-25'], + ['2026-03-25T23:59:59.999Z', '2026-03-25'], + ['2026-03-26T01:30:00.000+02:00', '2026-03-25'], + ['2026-03-25T20:30:00.000-05:00', '2026-03-26'], + ])('stores the UTC day of %s as the report date %s', async (now, reportDate) => { + const { id } = await createOn(now); + + await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({ reportDate }); + }); + + test('returns null for a second report on the same UTC day and keeps the first', async () => { + const first = await createOn('2026-03-25T07:42:00.000Z'); + + await expect( + repository.createPending(DATA_POINTS, new Date('2026-03-25T23:59:00.000Z')), + ).resolves.toBeNull(); + await expect(repository.find()).resolves.toEqual([expect.objectContaining({ id: first.id })]); + }); + + test('creates a report on the next UTC day', async () => { + await createOn('2026-03-25T23:59:00.000Z'); + + await createOn('2026-03-26T00:00:00.000Z'); + + await expect(repository.count()).resolves.toBe(2); + }); }); describe('findPending', () => { async function createdOn(date: string) { - const report = await repository.createPending(DATA_POINTS); + const report = await createOn(date); await repository.update({ id: report.id }, { createdAt: new Date(date) }); return report; @@ -55,7 +90,7 @@ describe('InstanceMonitoringReportRepository', () => { }); test('returns the pending report with the numbers it measured', async () => { - const created = await repository.createPending(DATA_POINTS); + const created = await createOn(new Date()); const pending = await repository.findPending(); @@ -65,7 +100,7 @@ describe('InstanceMonitoringReportRepository', () => { }); test('returns nothing once the newest report is delivered', async () => { - const created = await repository.createPending(DATA_POINTS); + const created = await createOn(new Date()); await repository.markDelivered(created.id, new Date()); await expect(repository.findPending()).resolves.toBeNull(); @@ -91,7 +126,7 @@ describe('InstanceMonitoringReportRepository', () => { }); test('carries the last attempt time, so the wait between attempts survives a restart', async () => { - const created = await repository.createPending(DATA_POINTS); + const created = await createOn(new Date()); const failedAt = new Date('2026-03-26T07:42:00.000Z'); await repository.recordFailure(created.id, 'Network error', failedAt); @@ -107,13 +142,13 @@ describe('InstanceMonitoringReportRepository', () => { }); test("is false while today's report is still pending", async () => { - await repository.createPending(DATA_POINTS); + await createOn(new Date()); await expect(repository.hasSettledToday(new Date())).resolves.toBe(false); }); test("is true once today's report was delivered", async () => { - const created = await repository.createPending(DATA_POINTS); + const created = await createOn(new Date()); await repository.markDelivered(created.id, new Date()); await expect(repository.hasSettledToday(new Date())).resolves.toBe(true); @@ -121,20 +156,32 @@ describe('InstanceMonitoringReportRepository', () => { test("is true once today's report ran out of attempts", async () => { // Otherwise the day would get a second report with a second budget. - const created = await repository.createPending(DATA_POINTS); + const created = await createOn(new Date()); await repository.markSkipped(created.id); await expect(repository.hasSettledToday(new Date())).resolves.toBe(true); }); test("ignores an earlier day's delivered report", async () => { - const delivered = await repository.createPending(DATA_POINTS); + const delivered = await createOn(new Date()); await repository.markDelivered(delivered.id, new Date()); const tomorrow = new Date(Date.now() + 24 * 60 * 60 * 1000); await expect(repository.hasSettledToday(tomorrow)).resolves.toBe(false); }); + + test('reads the UTC day the report was created on', async () => { + const delivered = await createOn('2026-03-25T23:59:00.000Z'); + await repository.markDelivered(delivered.id, new Date()); + + await expect(repository.hasSettledToday(new Date('2026-03-25T00:00:00.000Z'))).resolves.toBe( + true, + ); + await expect(repository.hasSettledToday(new Date('2026-03-26T00:01:00.000Z'))).resolves.toBe( + false, + ); + }); }); describe('findLastCoveredDay', () => { @@ -143,13 +190,13 @@ describe('InstanceMonitoringReportRepository', () => { } test('is null when nothing was ever delivered', async () => { - await repository.createPending(DATA_POINTS); + await createOn(new Date()); await expect(repository.findLastCoveredDay()).resolves.toBeNull(); }); test('returns the latest daily date a delivered report carried', async () => { - const delivered = await repository.createPending([ + const delivered = await createOn(new Date(), [ { kind: 'daily', name: 'billableExecutions', value: 1, date: '2026-03-24' }, { kind: 'daily', name: 'billableExecutions', value: 2, date: REPORT_DATE }, ]); @@ -162,12 +209,12 @@ describe('InstanceMonitoringReportRepository', () => { // The receiver answers 201 only once it saved the report, so a failure means // nothing was saved and the day is still owed. Reports pile up because the // scheduler creates a fresh one each day. - const delivered = await repository.createPending(daily('2026-03-22', 1)); + const delivered = await createOn('2026-03-23T07:42:00.000Z', daily('2026-03-22', 1)); await repository.markDelivered(delivered.id, new Date()); - const failed = await repository.createPending(daily('2026-03-23', 2)); + const failed = await createOn('2026-03-24T07:42:00.000Z', daily('2026-03-23', 2)); await repository.recordFailure(failed.id, 'Network error', new Date()); - await repository.createPending(daily('2026-03-24', 3)); + await createOn('2026-03-25T07:42:00.000Z', daily('2026-03-24', 3)); await expect(repository.findLastCoveredDay()).resolves.toBe('2026-03-22'); }); @@ -175,16 +222,16 @@ describe('InstanceMonitoringReportRepository', () => { describe('findLastDeliveryTime', () => { test('is null when nothing was ever delivered', async () => { - await repository.createPending(DATA_POINTS); + await createOn(new Date()); await expect(repository.findLastDeliveryTime()).resolves.toBeNull(); }); test('returns the latest delivery time across delivered reports', async () => { - const earlier = await repository.createPending(DATA_POINTS); + const earlier = await createOn('2026-03-24T07:41:00.000Z'); await repository.markDelivered(earlier.id, new Date('2026-03-24T07:42:00.000Z')); - const latest = await repository.createPending(DATA_POINTS); + const latest = await createOn('2026-03-25T07:42:00.000Z'); const latestDeliveredAt = new Date('2026-03-25T07:42:13.000Z'); await repository.markDelivered(latest.id, latestDeliveredAt); @@ -194,15 +241,15 @@ describe('InstanceMonitoringReportRepository', () => { }); test('ignores reports that never reached the receiver', async () => { - const delivered = await repository.createPending(DATA_POINTS); + const delivered = await createOn('2026-03-24T07:41:00.000Z'); const deliveredAt = new Date('2026-03-24T07:42:00.000Z'); await repository.markDelivered(delivered.id, deliveredAt); - const failed = await repository.createPending(DATA_POINTS); + const failed = await createOn('2026-03-25T07:42:00.000Z'); await repository.recordFailure(failed.id, 'Network error', new Date()); - const skipped = await repository.createPending(DATA_POINTS); + const skipped = await createOn('2026-03-26T07:42:00.000Z'); await repository.markSkipped(skipped.id); - await repository.createPending(DATA_POINTS); + await createOn('2026-03-27T07:42:00.000Z'); const lastDelivery = await repository.findLastDeliveryTime(); @@ -210,9 +257,32 @@ describe('InstanceMonitoringReportRepository', () => { }); }); + describe('markSkipped', () => { + test('settles a pending report as skipped', async () => { + const { id } = await createOn(new Date()); + + await repository.markSkipped(id); + + await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({ + status: 'skipped_after_max_retries', + }); + }); + + test('leaves a report delivered when another process delivered it first', async () => { + const { id } = await createOn(new Date()); + await repository.markDelivered(id, new Date()); + + await repository.markSkipped(id); + + await expect(repository.findOneByOrFail({ id })).resolves.toMatchObject({ + status: 'delivered', + }); + }); + }); + describe('markDelivered', () => { test('stamps the delivery time, counts the attempt and clears any earlier error', async () => { - const { id } = await repository.createPending(DATA_POINTS); + const { id } = await createOn(new Date()); await repository.recordFailure(id, 'Network error', new Date()); const deliveredAt = new Date('2026-03-26T07:42:00.000Z'); @@ -227,7 +297,7 @@ describe('InstanceMonitoringReportRepository', () => { describe('recordFailure', () => { test('counts the attempt and keeps the report undelivered', async () => { - const { id } = await repository.createPending(DATA_POINTS); + const { id } = await createOn(new Date()); await repository.recordFailure(id, 'Network error', new Date()); await repository.recordFailure(id, 'Still down', new Date()); diff --git a/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts b/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts index 54f05caf52e..258f8cf46f2 100644 --- a/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts +++ b/packages/cli/src/modules/instance-reporting.ee/database/repositories/instance-monitoring-report.repository.ts @@ -1,5 +1,6 @@ +import { isUniqueConstraintError } from '@n8n/db'; import { Service } from '@n8n/di'; -import { DataSource, IsNull, MoreThanOrEqual, Not, Repository } from '@n8n/typeorm'; +import { DataSource, IsNull, Not, Repository } from '@n8n/typeorm'; import { v4 as uuid } from 'uuid'; import type { InstanceReportDataPoint } from '../entities/instance-monitoring-report'; @@ -47,20 +48,33 @@ export class InstanceMonitoringReportRepository extends Repository { - return await this.existsBy({ - createdAt: MoreThanOrEqual(startOfUtcDay(now)), - status: Not('pending'), - }); + return await this.existsBy({ reportDate: utcDay(now), status: Not('pending') }); } /** * Record a freshly measured report, with its data points, before any attempt * to deliver it. A row therefore always carries the measurement it stands for. + * + * `null` when a report was already created on `now`'s UTC day. */ - async createPending(dataPoints: InstanceReportDataPoint[]): Promise { - return await this.save( - this.create({ id: uuid(), dataPoints, status: 'pending', deliveredAt: null }), - ); + async createPending( + dataPoints: InstanceReportDataPoint[], + now: Date, + ): Promise { + try { + return await this.save( + this.create({ + id: uuid(), + reportDate: utcDay(now), + dataPoints, + status: 'pending', + deliveredAt: null, + }), + ); + } catch (error) { + if (isUniqueConstraintError(error)) return null; + throw error; + } } /** @@ -112,14 +126,15 @@ export class InstanceMonitoringReportRepository extends Repository { - await this.update({ id }, { status: 'skipped_after_max_retries' }); + await this.update({ id, status: 'pending' }, { status: 'skipped_after_max_retries' }); } } -function startOfUtcDay(instant: Date): Date { - return new Date( - Date.UTC(instant.getUTCFullYear(), instant.getUTCMonth(), instant.getUTCDate(), 0, 0, 0, 0), - ); +function utcDay(instant: Date): string { + return instant.toISOString().slice(0, 10); } diff --git a/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts b/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts index 293e504a509..92e75fcaf25 100644 --- a/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts +++ b/packages/cli/src/modules/instance-reporting.ee/instance-reporting-scheduler.service.ts @@ -10,10 +10,16 @@ import { EventService } from '@/events/event.service'; import type { InstanceMonitoringReport } from './database/entities/instance-monitoring-report'; import { InstanceMonitoringReportRepository } from './database/repositories/instance-monitoring-report.repository'; import { InstanceReportingSettingsService } from './instance-reporting-settings.service'; -import { InstanceReportingService, RETRY_DELAY_MS } from './instance-reporting.service'; +import { + InstanceReportAlreadyCreatedError, + InstanceReportingService, +} from './instance-reporting.service'; const MINUTES_PER_DAY = 24 * 60; +/** How long to wait before re-attempting a delivery that failed. */ +const RETRY_DELAY_MS = 5 * Time.minutes.toMilliseconds; + /** * Fires the daily instance report at this instance's configured report time. * @@ -118,7 +124,7 @@ export class InstanceReportingScheduler { // the slot collapses to zero, so this pass falls through to reportIfDue, // which skips the stale row and sends a fresh report. const waitMs = Math.min( - await this.reportingService.msUntilRetryAllowed(new Date()), + await this.msUntilRetryAllowed(new Date()), await this.msUntilNextSlot(reportTime), ); if (waitMs > 0) { @@ -195,6 +201,18 @@ export class InstanceReportingScheduler { await this.reportingService.sendReport(); return 'sent'; } catch (error) { + if (error instanceof InstanceReportAlreadyCreatedError) { + this.logger.info( + 'Unexpected second instance report triggered for today, ignored because there is already a report for today', + ); + // In this case, the report is not actually considered "failed" from a business perspective, + // but our code has a bug that led to a report being generated a second time for the same day. + // Returning 'failed' here, will lead to a retry of the report being scheduled after RETRY_DELAY_MS. + // The expectation is that after that single no-op retry, retrying stops + // as InstanceReportingService.sendReport will exit early. + return 'failed'; + } + this.logger.warn('Failed to deliver the instance report', { error }); return 'failed'; } @@ -205,6 +223,23 @@ export class InstanceReportingScheduler { this.timeout = setTimeout(async () => await this.tick(), delayMs); } + + /** + * How long the scheduler must wait before attempting today's report again, or + * `0` when it may attempt now. + * + * Derived from the report row, so the wait survives a restart. Without it, a + * crash loop would attempt at once every time and spend the whole budget in + * seconds. + */ + private async msUntilRetryAllowed(now: Date): Promise { + const pending = await this.reportRepository.findPending(); + if (!pending?.lastAttemptAt) return 0; + + const elapsed = now.getTime() - pending.lastAttemptAt.getTime(); + + return Math.max(0, RETRY_DELAY_MS - elapsed); + } } /** diff --git a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts index 56aac20325e..94e465c56f7 100644 --- a/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts +++ b/packages/cli/src/modules/instance-reporting.ee/instance-reporting.service.ts @@ -39,6 +39,12 @@ class InstanceReportRejectedError extends OperationalError { } } +export class InstanceReportAlreadyCreatedError extends OperationalError { + constructor() { + super('Another process already created the instance report for today'); + } +} + type SkipReason = 'max-retries' | 'slot-passed' | 'rejected'; const SKIP_MESSAGES: Record = { @@ -47,9 +53,6 @@ const SKIP_MESSAGES: Record = { rejected: 'Giving up on the instance report because the receiver rejected its payload', }; -/** How long to wait before re-attempting a delivery that failed. */ -export const RETRY_DELAY_MS = 5 * Time.minutes.toMilliseconds; - /** * How many missed days one report may carry. * @@ -120,7 +123,8 @@ export class InstanceReportingService { * * A 400 or 413 skips the report at once, since a resend carries the same payload. * - * @throws when delivery fails and a retry may succeed, so the scheduler retries with backoff. + * @throws when delivery fails and a retry may succeed, or another process created + * today's report, so the scheduler retries with backoff. */ async sendReport(): Promise { const licenseCert = this.config.instanceReportingAuthToken @@ -154,7 +158,10 @@ export class InstanceReportingService { }); } - report = await this.reportRepository.createPending(await this.collectDataPoints(days)); + report = await this.reportRepository.createPending(await this.collectDataPoints(days), now); + if (!report) { + throw new InstanceReportAlreadyCreatedError(); + } } const payload = { @@ -232,23 +239,6 @@ export class InstanceReportingService { this.logger.error(SKIP_MESSAGES[reason], { batchId: id, attempts, lastError }); } - /** - * How long the scheduler must wait before attempting today's report again, or - * `0` when it may attempt now. - * - * Derived from the report row, so the wait survives a restart. Without it, a - * crash loop would attempt at once every time and spend the whole budget in - * seconds. - */ - async msUntilRetryAllowed(now: Date): Promise { - const pending = await this.reportRepository.findPending(); - if (!pending?.lastAttemptAt) return 0; - - const elapsed = now.getTime() - pending.lastAttemptAt.getTime(); - - return Math.max(0, RETRY_DELAY_MS - elapsed); - } - /** * The UTC days this report must carry a daily point for, oldest first, ending * yesterday. Empty when yesterday is already reported. diff --git a/packages/cli/test/migration/1790583238183-add-report-date-to-instance-monitoring-report.test.ts b/packages/cli/test/migration/1790583238183-add-report-date-to-instance-monitoring-report.test.ts new file mode 100644 index 00000000000..d87074807d1 --- /dev/null +++ b/packages/cli/test/migration/1790583238183-add-report-date-to-instance-monitoring-report.test.ts @@ -0,0 +1,253 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + undoLastSingleMigration, + type TestMigrationContext, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; +import { randomUUID } from 'node:crypto'; + +const MIGRATION_NAME = 'AddReportDateToInstanceMonitoringReport1790583238183'; +const TABLE = 'instance_monitoring_report'; + +type Status = 'pending' | 'delivered' | 'skipped_after_max_retries'; + +describe('AddReportDateToInstanceMonitoringReport Migration', () => { + let dataSource: DataSource; + + beforeAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.init(); + dataSource = Container.get(DataSource); + }); + + beforeEach(async () => { + const context = createTestMigrationContext(dataSource); + await context.queryRunner.clearDatabase(); + await context.queryRunner.release(); + await initDbUpToMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.close(); + }); + + async function withContext(run: (context: TestMigrationContext) => Promise): Promise { + const context = createTestMigrationContext(dataSource); + try { + return await run(context); + } finally { + await context.queryRunner.release(); + } + } + + async function insertReport( + context: TestMigrationContext, + { createdAt, status = 'delivered' }: { createdAt: string; status?: Status }, + ): Promise { + const id = randomUUID(); + await context.runQuery( + `INSERT INTO ${context.escape.tableName(TABLE)} + ("id", "dataPoints", "status", "createdAt", "updatedAt") + VALUES (:id, '[]', :status, :createdAt, :createdAt)`, + { id, status, createdAt }, + ); + return id; + } + + async function insertReportOn(context: TestMigrationContext, reportDate: string) { + await context.runQuery( + `INSERT INTO ${context.escape.tableName(TABLE)} + ("id", "dataPoints", "reportDate", "createdAt", "updatedAt") + VALUES (:id, '[]', :reportDate, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`, + { id: randomUUID(), reportDate }, + ); + } + + async function reportDates( + context: TestMigrationContext, + ): Promise> { + const rows = (await context.queryRunner.query( + `SELECT "id", "reportDate" FROM ${context.escape.tableName(TABLE)}`, + )) as Array<{ id: string; reportDate: string | null }>; + return Object.fromEntries(rows.map((row) => [row.id, row.reportDate])); + } + + async function statuses(context: TestMigrationContext): Promise> { + const rows = (await context.queryRunner.query( + `SELECT "id", "status" FROM ${context.escape.tableName(TABLE)}`, + )) as Array<{ id: string; status: Status }>; + return Object.fromEntries(rows.map((row) => [row.id, row.status])); + } + + async function columnNames(context: TestMigrationContext): Promise { + if (context.isSqlite) { + const rows = (await context.queryRunner.query( + `PRAGMA table_info(${context.escape.tableName(TABLE)})`, + )) as Array<{ name: string }>; + return rows.map((row) => row.name); + } + const rows = (await context.queryRunner.query( + 'SELECT column_name FROM information_schema.columns WHERE table_name = $1', + [`${context.tablePrefix}${TABLE}`], + )) as Array<{ column_name: string }>; + return rows.map((row) => row.column_name); + } + + async function indexExists(context: TestMigrationContext): Promise { + const name = `IDX_${context.tablePrefix}${TABLE}_reportDate`; + const rows = context.isSqlite + ? await context.runQuery( + "SELECT name FROM sqlite_master WHERE type = 'index' AND name = :name", + { name }, + ) + : await context.runQuery( + 'SELECT indexname FROM pg_indexes WHERE indexname = :name', + { name }, + ); + return rows.length === 1; + } + + describe('up', () => { + it('backfills the UTC day each report was created on', async () => { + const ids = await withContext(async (context) => ({ + morning: await insertReport(context, { createdAt: '2026-03-25 07:42:00.000+00:00' }), + eastOfUtc: await insertReport(context, { createdAt: '2026-03-27 01:30:00.000+02:00' }), + beforeMidnight: await insertReport(context, { createdAt: '2026-03-27 23:59:59.999+00:00' }), + })); + + await runSingleMigration(MIGRATION_NAME); + + await expect(withContext(reportDates)).resolves.toEqual({ + [ids.morning]: '2026-03-25', + [ids.eastOfUtc]: '2026-03-26', + [ids.beforeMidnight]: '2026-03-27', + }); + }); + + it('keeps the day on the delivered row, then the skipped one, when several share it', async () => { + const ids = await withContext(async (context) => ({ + pending: await insertReport(context, { + createdAt: '2026-03-25 09:00:00.000+00:00', + status: 'pending', + }), + skipped: await insertReport(context, { + createdAt: '2026-03-25 08:00:00.000+00:00', + status: 'skipped_after_max_retries', + }), + delivered: await insertReport(context, { createdAt: '2026-03-25 07:00:00.000+00:00' }), + skippedAlone: await insertReport(context, { + createdAt: '2026-03-26 08:00:00.000+00:00', + status: 'skipped_after_max_retries', + }), + pendingNextToSkipped: await insertReport(context, { + createdAt: '2026-03-26 09:00:00.000+00:00', + status: 'pending', + }), + })); + + await runSingleMigration(MIGRATION_NAME); + + await expect(withContext(reportDates)).resolves.toEqual({ + [ids.pending]: null, + [ids.skipped]: null, + [ids.delivered]: '2026-03-25', + [ids.skippedAlone]: '2026-03-26', + [ids.pendingNextToSkipped]: null, + }); + }); + + it('settles a pending row that loses its day, so it is not sent again', async () => { + const ids = await withContext(async (context) => ({ + delivered: await insertReport(context, { createdAt: '2026-03-25 07:00:00.000+00:00' }), + pending: await insertReport(context, { + createdAt: '2026-03-25 09:00:00.000+00:00', + status: 'pending', + }), + pendingAlone: await insertReport(context, { + createdAt: '2026-03-26 09:00:00.000+00:00', + status: 'pending', + }), + })); + + await runSingleMigration(MIGRATION_NAME); + + await expect(withContext(statuses)).resolves.toEqual({ + [ids.delivered]: 'delivered', + [ids.pending]: 'skipped_after_max_retries', + [ids.pendingAlone]: 'pending', + }); + }); + + it('keeps the day on the newest row when rows with one status share it', async () => { + const ids = await withContext(async (context) => ({ + older: await insertReport(context, { createdAt: '2026-03-25 07:00:00.000+00:00' }), + newer: await insertReport(context, { createdAt: '2026-03-25 08:00:00.000+00:00' }), + })); + + await runSingleMigration(MIGRATION_NAME); + + await expect(withContext(reportDates)).resolves.toEqual({ + [ids.older]: null, + [ids.newer]: '2026-03-25', + }); + }); + + it('rejects a second report on a day that already has one', async () => { + await runSingleMigration(MIGRATION_NAME); + + await withContext(async (context) => { + expect(await indexExists(context)).toBe(true); + await insertReportOn(context, '2026-03-25'); + + await expect(insertReportOn(context, '2026-03-25')).rejects.toThrow(); + }); + }); + + it('accepts reports on other days and legacy rows without a day', async () => { + await withContext(async (context) => { + await insertReport(context, { createdAt: '2026-03-25 07:00:00.000+00:00' }); + await insertReport(context, { createdAt: '2026-03-25 08:00:00.000+00:00' }); + await insertReport(context, { createdAt: '2026-03-25 09:00:00.000+00:00' }); + }); + + await runSingleMigration(MIGRATION_NAME); + + await withContext(async (context) => { + await insertReportOn(context, '2026-03-26'); + + expect(Object.values(await reportDates(context)).sort()).toEqual([ + '2026-03-25', + '2026-03-26', + null, + null, + ]); + }); + }); + }); + + describe('down', () => { + it('drops the column and its index without deleting a report', async () => { + await withContext(async (context) => { + await insertReport(context, { createdAt: '2026-03-25 07:00:00.000+00:00' }); + await insertReport(context, { createdAt: '2026-03-25 08:00:00.000+00:00' }); + }); + await runSingleMigration(MIGRATION_NAME); + + await undoLastSingleMigration(); + + await withContext(async (context) => { + expect(await indexExists(context)).toBe(false); + expect(await columnNames(context)).not.toContain('reportDate'); + const [{ count }] = (await context.queryRunner.query( + `SELECT COUNT(*) AS "count" FROM ${context.escape.tableName(TABLE)}`, + )) as Array<{ count: number | string }>; + expect(Number(count)).toBe(2); + }); + }); + }); +});