From 0bb5e342183e5cd0cc226bfbb58ad3ff84a917d3 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 8 Oct 2026 22:00:11 +0200 Subject: [PATCH 1/3] feat: drop shadow-diff body mismatches on records edited after capture (CM-1473) Signed-off-by: Mouad BANI --- .../src/activities/shadowDiffActivities.ts | 28 ++- .../src/editedRecords.test.ts | 164 ++++++++++++++++++ .../connectors_worker/src/editedRecords.ts | 133 ++++++++++++++ 3 files changed, 324 insertions(+), 1 deletion(-) create mode 100644 services/apps/connectors_worker/src/editedRecords.test.ts create mode 100644 services/apps/connectors_worker/src/editedRecords.ts diff --git a/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts b/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts index fa8915419c..5f809f966a 100644 --- a/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts +++ b/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts @@ -16,6 +16,7 @@ import { hasDeletedRecordCandidates, isDeletedRecordCandidate, } from '../deletedRecords' +import { dropConfirmedEditedRecords, hasEditedRecordCandidates } from '../editedRecords' import { dropConfirmedForcePushedCommits, hasForcePushCandidates, @@ -180,8 +181,9 @@ export async function runShadowDiffForChannel( const needsForcePushCheck = hasForcePushCandidates(unit.syncName, result.mismatches) const needsDeletedRecordCheck = hasDeletedRecordCandidates(unit.syncName, result.mismatches) + const needsEditedRecordCheck = hasEditedRecordCandidates(unit.syncName, result.mismatches) - if (needsForcePushCheck || needsDeletedRecordCheck) { + if (needsForcePushCheck || needsDeletedRecordCheck || needsEditedRecordCheck) { confirmationHttp ??= createGithubConfirmationHttp(qx, channel.integrationId) let http: ConnectorHttp try { @@ -256,6 +258,30 @@ export async function runShadowDiffForChannel( } result = { ...result, mismatches } } + + if (hasEditedRecordCandidates(unit.syncName, result.mismatches)) { + const { mismatches, confirmedEditedCount, unconfirmedCount } = + await dropConfirmedEditedRecords(unit.syncName, result.mismatches, http, svc.log) + if (confirmedEditedCount > 0) { + svc.log.info( + { unitId: unit.id, day, channelName: channel.channelName, confirmedEditedCount }, + 'skipped body mismatches on records edited after capture', + ) + } + if (unconfirmedCount > 0) { + svc.log.warn( + { + unitId: unit.id, + day, + channelName: channel.channelName, + syncName: unit.syncName, + checkFailedCount: unconfirmedCount, + }, + 'shadow diff check_failed: edited-record candidates not confirmed against github, kept as field_mismatch', + ) + } + result = { ...result, mismatches } + } } unitDiffs.push({ unit, result }) diff --git a/services/apps/connectors_worker/src/editedRecords.test.ts b/services/apps/connectors_worker/src/editedRecords.test.ts new file mode 100644 index 0000000000..780350bc26 --- /dev/null +++ b/services/apps/connectors_worker/src/editedRecords.test.ts @@ -0,0 +1,164 @@ +import { describe, expect, it, vi } from 'vitest' + +import type { ConnectorHttp } from '@crowd/connectors' +import type { Logger } from '@crowd/logging' + +import { dropConfirmedEditedRecords, hasEditedRecordCandidates } from './editedRecords' +import { IShadowDiffMismatch } from './shadowDiff' + +const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as unknown as Logger + +function bodyMismatch(sourceId: string, type = 'issue-comment'): IShadowDiffMismatch { + return { + sourceId, + type, + kind: 'field_mismatch', + severity: 'high', + fields: [{ field: 'body', shadowValue: 'old', nangoValue: 'new' }], + } +} + +function httpWithHandler(handler: (ids: string[]) => Promise): ConnectorHttp { + return { + request: async (config: { data: { variables: { ids: string[] } } }) => + handler(config.data.variables.ids), + requestCount: () => 0, + } as unknown as ConnectorHttp +} + +function editedResponse(editedIds: string[]) { + return async (ids: string[]) => ({ + data: { + nodes: ids.map((id) => ({ + id, + lastEditedAt: editedIds.includes(id) ? '2026-10-07T00:00:00Z' : null, + })), + }, + }) +} + +describe('hasEditedRecordCandidates', () => { + it('is false for the pull-request-commits sync', () => { + expect( + hasEditedRecordCandidates('pull-request-commits', [ + bodyMismatch('abc123', 'authored-commit'), + ]), + ).toBe(false) + }) + + it('is false when the mismatch involves fields other than body', () => { + const mismatch: IShadowDiffMismatch = { + ...bodyMismatch('IC_abc'), + fields: [ + { field: 'body', shadowValue: 'old', nangoValue: 'new' }, + { field: 'url', shadowValue: 'a', nangoValue: 'b' }, + ], + } + expect(hasEditedRecordCandidates('issue-comments', [mismatch])).toBe(false) + }) + + it('is false for missing_in_shadow mismatches', () => { + expect( + hasEditedRecordCandidates('issue-comments', [ + { sourceId: 'IC_abc', type: 'issue-comment', kind: 'missing_in_shadow', severity: 'high' }, + ]), + ).toBe(false) + }) + + it('is true for a body-only mismatch on a real node id', () => { + expect(hasEditedRecordCandidates('issue-comments', [bodyMismatch('IC_abc')])).toBe(true) + }) + + it('is true for a body-only mismatch on a synthetic id carrying a parent node', () => { + expect( + hasEditedRecordCandidates('pull-requests', [ + bodyMismatch( + 'gen-RRE_PR_kwDOI7xefs8AAAABGaViEg_alice_bob_2026-10-06T15:07:23.000Z', + 'pull_request-review-requested', + ), + ]), + ).toBe(true) + }) +}) + +describe('dropConfirmedEditedRecords', () => { + it('drops body mismatches whose node was edited and keeps the never-edited ones', async () => { + const mismatches = [bodyMismatch('IC_edited'), bodyMismatch('IC_pristine')] + const http = httpWithHandler(editedResponse(['IC_edited'])) + + const result = await dropConfirmedEditedRecords('issue-comments', mismatches, http, log) + + expect(result.mismatches).toEqual([bodyMismatch('IC_pristine')]) + expect(result.confirmedEditedCount).toBe(1) + expect(result.unconfirmedCount).toBe(0) + }) + + it('resolves synthetic timeline ids to their parent node and queries each parent once', async () => { + const prId = 'PR_kwDOI7xefs8AAAABGaViEg' + const mismatches = [ + bodyMismatch( + `gen-RRE_${prId}_alice_bob_2026-10-06T15:07:23.000Z`, + 'pull_request-review-requested', + ), + bodyMismatch( + `gen-RRE_${prId}_alice_carol_2026-10-06T15:07:23.000Z`, + 'pull_request-review-requested', + ), + bodyMismatch(prId, 'pull_request-opened'), + ] + const seen: string[][] = [] + const http = httpWithHandler(async (ids) => { + seen.push(ids) + return editedResponse([prId])(ids) + }) + + const result = await dropConfirmedEditedRecords('pull-requests', mismatches, http, log) + + expect(seen).toEqual([[prId]]) + expect(result.mismatches).toEqual([]) + expect(result.confirmedEditedCount).toBe(3) + }) + + it('keeps mismatches and counts them unconfirmed when the request fails', async () => { + const mismatches = [bodyMismatch('IC_unknown')] + const http = httpWithHandler(async () => { + throw new Error('network exploded') + }) + + const result = await dropConfirmedEditedRecords('issue-comments', mismatches, http, log) + + expect(result.mismatches).toEqual(mismatches) + expect(result.confirmedEditedCount).toBe(0) + expect(result.unconfirmedCount).toBe(1) + }) + + it('keeps mismatches whose node resolved to null', async () => { + const mismatches = [bodyMismatch('IC_gone')] + const http = httpWithHandler(async (ids) => ({ data: { nodes: ids.map(() => null) } })) + + const result = await dropConfirmedEditedRecords('issue-comments', mismatches, http, log) + + expect(result.mismatches).toEqual(mismatches) + expect(result.confirmedEditedCount).toBe(0) + expect(result.unconfirmedCount).toBe(0) + }) + + it('leaves non-candidate mismatches untouched', async () => { + const other: IShadowDiffMismatch = { + sourceId: 'IC_other', + type: 'issue-comment', + kind: 'missing_in_nango', + severity: 'high', + } + const http = httpWithHandler(editedResponse(['IC_edited'])) + + const result = await dropConfirmedEditedRecords( + 'issue-comments', + [other, bodyMismatch('IC_edited')], + http, + log, + ) + + expect(result.mismatches).toEqual([other]) + }) +}) diff --git a/services/apps/connectors_worker/src/editedRecords.ts b/services/apps/connectors_worker/src/editedRecords.ts new file mode 100644 index 0000000000..f8b384a114 --- /dev/null +++ b/services/apps/connectors_worker/src/editedRecords.ts @@ -0,0 +1,133 @@ +import type { ConnectorHttp } from '@crowd/connectors' +import { mapWithConcurrency } from '@crowd/connectors' +import type { Logger } from '@crowd/logging' + +import { IShadowDiffMismatch } from './shadowDiff' + +const COMMIT_SYNC_NAME = 'pull-request-commits' +const SYNTHETIC_SOURCE_ID_NODE = /^gen-[A-Z]+_([A-Z]+_[A-Za-z0-9-]+)(?:_|$)/ +const NODES_QUERY_BATCH_SIZE = 100 +const CONFIRM_CONCURRENCY = 3 +const CONFIRM_BUDGET_MS = 60_000 +const CONFIRM_REQUEST_TIMEOUT_MS = 10_000 +const CONFIRM_REQUEST_MAX_ATTEMPTS = 1 + +const NODES_QUERY = `query($ids: [ID!]!) { nodes(ids: $ids) { id ... on Comment { lastEditedAt } } }` + +interface GraphqlEnvelope { + data?: T +} + +interface NodesQueryResult { + nodes: ({ id: string; lastEditedAt?: string | null } | null)[] +} + +function editedNodeId(mismatch: IShadowDiffMismatch): string | null { + const synthetic = SYNTHETIC_SOURCE_ID_NODE.exec(mismatch.sourceId) + if (synthetic) { + return synthetic[1] + } + return mismatch.sourceId.startsWith('gen-') ? null : mismatch.sourceId +} + +export function isEditedRecordCandidate(syncName: string, mismatch: IShadowDiffMismatch): boolean { + return ( + syncName !== COMMIT_SYNC_NAME && + mismatch.kind === 'field_mismatch' && + (mismatch.fields?.length ?? 0) > 0 && + mismatch.fields.every((f) => f.field === 'body') && + editedNodeId(mismatch) !== null + ) +} + +export function hasEditedRecordCandidates( + syncName: string, + mismatches: IShadowDiffMismatch[], +): boolean { + return mismatches.some((m) => isEditedRecordCandidate(syncName, m)) +} + +export interface EditedRecordFilterResult { + mismatches: IShadowDiffMismatch[] + confirmedEditedCount: number + unconfirmedCount: number +} + +function toBatches(items: T[], size: number): T[][] { + const batches: T[][] = [] + for (let i = 0; i < items.length; i += size) { + batches.push(items.slice(i, i + size)) + } + return batches +} + +async function fetchEditedNodeIds( + http: ConnectorHttp, + ids: string[], + log: Logger, +): Promise | null> { + try { + const body = await http.request>( + { + method: 'post', + url: 'https://api.github.com/graphql', + data: { query: NODES_QUERY, variables: { ids } }, + timeout: CONFIRM_REQUEST_TIMEOUT_MS, + }, + log, + CONFIRM_REQUEST_MAX_ATTEMPTS, + ) + if (!body.data || body.data.nodes.length !== ids.length) { + return null + } + return new Set( + body.data.nodes + .filter((node): node is { id: string; lastEditedAt: string } => Boolean(node?.lastEditedAt)) + .map((node) => node.id), + ) + } catch { + return null + } +} + +export async function dropConfirmedEditedRecords( + syncName: string, + mismatches: IShadowDiffMismatch[], + http: ConnectorHttp, + log: Logger, +): Promise { + const candidates = mismatches.filter((m) => isEditedRecordCandidate(syncName, m)) + if (candidates.length === 0) { + return { mismatches, confirmedEditedCount: 0, unconfirmedCount: 0 } + } + + const nodeIds = [...new Set(candidates.map((m) => editedNodeId(m) as string))] + const batches = toBatches(nodeIds, NODES_QUERY_BATCH_SIZE) + const deadline = Date.now() + CONFIRM_BUDGET_MS + + const editedNodeIds = new Set() + const uncheckedNodeIds = new Set() + + await mapWithConcurrency(batches, CONFIRM_CONCURRENCY, async (batch) => { + if (Date.now() >= deadline) { + for (const id of batch) uncheckedNodeIds.add(id) + return + } + const edited = await fetchEditedNodeIds(http, batch, log) + if (edited === null) { + for (const id of batch) uncheckedNodeIds.add(id) + return + } + for (const id of edited) editedNodeIds.add(id) + }) + + const isConfirmedEdited = (m: IShadowDiffMismatch) => + isEditedRecordCandidate(syncName, m) && editedNodeIds.has(editedNodeId(m) as string) + + return { + mismatches: mismatches.filter((m) => !isConfirmedEdited(m)), + confirmedEditedCount: candidates.filter(isConfirmedEdited).length, + unconfirmedCount: candidates.filter((m) => uncheckedNodeIds.has(editedNodeId(m) as string)) + .length, + } +} From 8b0351e1c5b8653eda23b6fe63b3c915a8c56090 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 8 Oct 2026 22:05:43 +0200 Subject: [PATCH 2/3] fix: scope edited-record check to parent-body events, log unchecked ones (CM-1473) Signed-off-by: Mouad BANI --- .../src/activities/shadowDiffActivities.ts | 14 ++++++++++---- .../connectors_worker/src/editedRecords.test.ts | 11 +++++++++++ .../apps/connectors_worker/src/editedRecords.ts | 4 ++-- 3 files changed, 23 insertions(+), 6 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts b/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts index 5f809f966a..434fac681e 100644 --- a/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts +++ b/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts @@ -16,7 +16,11 @@ import { hasDeletedRecordCandidates, isDeletedRecordCandidate, } from '../deletedRecords' -import { dropConfirmedEditedRecords, hasEditedRecordCandidates } from '../editedRecords' +import { + dropConfirmedEditedRecords, + hasEditedRecordCandidates, + isEditedRecordCandidate, +} from '../editedRecords' import { dropConfirmedForcePushedCommits, hasForcePushCandidates, @@ -98,13 +102,14 @@ export async function listShadowDiffChannels(): Promise { function dropUncheckedCandidates( syncName: string, result: IShadowDiffUnitResult, -): { result: IShadowDiffUnitResult; droppedCount: number } { +): { result: IShadowDiffUnitResult; droppedCount: number; keptEditedCount: number } { const isCandidate = (m: IShadowDiffMismatch) => isForcePushCandidate(syncName, m) || isDeletedRecordCandidate(syncName, m) const mismatches = result.mismatches.filter((m) => !isCandidate(m)) return { result: { ...result, mismatches }, droppedCount: result.mismatches.length - mismatches.length, + keptEditedCount: mismatches.filter((m) => isEditedRecordCandidate(syncName, m)).length, } } @@ -197,9 +202,10 @@ export async function runShadowDiffForChannel( day, channelName: channel.channelName, syncName: unit.syncName, - checkFailedCount: dropped.droppedCount, + checkFailedCount: dropped.droppedCount + dropped.keptEditedCount, + uncheckedEditedCount: dropped.keptEditedCount, }, - 'shadow diff check_failed: github client setup failed, candidates excluded from missing_in_shadow', + 'shadow diff check_failed: github client setup failed, candidates excluded from missing_in_shadow and edited-record candidates kept as field_mismatch', ) unitDiffs.push({ unit, result: dropped.result }) continue diff --git a/services/apps/connectors_worker/src/editedRecords.test.ts b/services/apps/connectors_worker/src/editedRecords.test.ts index 780350bc26..23ae9b23bf 100644 --- a/services/apps/connectors_worker/src/editedRecords.test.ts +++ b/services/apps/connectors_worker/src/editedRecords.test.ts @@ -79,6 +79,17 @@ describe('hasEditedRecordCandidates', () => { ]), ).toBe(true) }) + + it('is false for a reviewed event whose body belongs to the review, not the parent PR', () => { + expect( + hasEditedRecordCandidates('pull-requests', [ + bodyMismatch( + 'gen-PRR_PR_kwDOI7xefs8AAAABGaViEg_alice_2026-10-06T15:07:23.000Z', + 'pull_request-reviewed', + ), + ]), + ).toBe(false) + }) }) describe('dropConfirmedEditedRecords', () => { diff --git a/services/apps/connectors_worker/src/editedRecords.ts b/services/apps/connectors_worker/src/editedRecords.ts index f8b384a114..6fb944c794 100644 --- a/services/apps/connectors_worker/src/editedRecords.ts +++ b/services/apps/connectors_worker/src/editedRecords.ts @@ -5,7 +5,7 @@ import type { Logger } from '@crowd/logging' import { IShadowDiffMismatch } from './shadowDiff' const COMMIT_SYNC_NAME = 'pull-request-commits' -const SYNTHETIC_SOURCE_ID_NODE = /^gen-[A-Z]+_([A-Z]+_[A-Za-z0-9-]+)(?:_|$)/ +const PARENT_BODY_SYNTHETIC_SOURCE_ID = /^gen-(?:AE|CE|ME|RRE)_([A-Z]+_[A-Za-z0-9-]+)(?:_|$)/ const NODES_QUERY_BATCH_SIZE = 100 const CONFIRM_CONCURRENCY = 3 const CONFIRM_BUDGET_MS = 60_000 @@ -23,7 +23,7 @@ interface NodesQueryResult { } function editedNodeId(mismatch: IShadowDiffMismatch): string | null { - const synthetic = SYNTHETIC_SOURCE_ID_NODE.exec(mismatch.sourceId) + const synthetic = PARENT_BODY_SYNTHETIC_SOURCE_ID.exec(mismatch.sourceId) if (synthetic) { return synthetic[1] } From 95e3b1c6bef527bd02173305f7ce3b5a71ba98dd Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 8 Oct 2026 22:31:32 +0200 Subject: [PATCH 3/3] fix: count null nodes as unchecked in the edited-record confirmation (CM-1473) Signed-off-by: Mouad BANI --- .../src/editedRecords.test.ts | 4 +-- .../connectors_worker/src/editedRecords.ts | 29 +++++++++++++------ 2 files changed, 22 insertions(+), 11 deletions(-) diff --git a/services/apps/connectors_worker/src/editedRecords.test.ts b/services/apps/connectors_worker/src/editedRecords.test.ts index 23ae9b23bf..abdec57a1b 100644 --- a/services/apps/connectors_worker/src/editedRecords.test.ts +++ b/services/apps/connectors_worker/src/editedRecords.test.ts @@ -143,7 +143,7 @@ describe('dropConfirmedEditedRecords', () => { expect(result.unconfirmedCount).toBe(1) }) - it('keeps mismatches whose node resolved to null', async () => { + it('keeps mismatches whose node resolved to null and counts them unconfirmed', async () => { const mismatches = [bodyMismatch('IC_gone')] const http = httpWithHandler(async (ids) => ({ data: { nodes: ids.map(() => null) } })) @@ -151,7 +151,7 @@ describe('dropConfirmedEditedRecords', () => { expect(result.mismatches).toEqual(mismatches) expect(result.confirmedEditedCount).toBe(0) - expect(result.unconfirmedCount).toBe(0) + expect(result.unconfirmedCount).toBe(1) }) it('leaves non-candidate mismatches untouched', async () => { diff --git a/services/apps/connectors_worker/src/editedRecords.ts b/services/apps/connectors_worker/src/editedRecords.ts index 6fb944c794..630e949c36 100644 --- a/services/apps/connectors_worker/src/editedRecords.ts +++ b/services/apps/connectors_worker/src/editedRecords.ts @@ -61,11 +61,16 @@ function toBatches(items: T[], size: number): T[][] { return batches } +interface BatchConfirmation { + editedIds: Set + unresolvedIds: Set +} + async function fetchEditedNodeIds( http: ConnectorHttp, ids: string[], log: Logger, -): Promise | null> { +): Promise { try { const body = await http.request>( { @@ -80,11 +85,16 @@ async function fetchEditedNodeIds( if (!body.data || body.data.nodes.length !== ids.length) { return null } - return new Set( - body.data.nodes - .filter((node): node is { id: string; lastEditedAt: string } => Boolean(node?.lastEditedAt)) - .map((node) => node.id), - ) + const editedIds = new Set() + const unresolvedIds = new Set() + body.data.nodes.forEach((node, i) => { + if (node === null) { + unresolvedIds.add(ids[i]) + } else if (node.lastEditedAt) { + editedIds.add(node.id) + } + }) + return { editedIds, unresolvedIds } } catch { return null } @@ -113,12 +123,13 @@ export async function dropConfirmedEditedRecords( for (const id of batch) uncheckedNodeIds.add(id) return } - const edited = await fetchEditedNodeIds(http, batch, log) - if (edited === null) { + const confirmation = await fetchEditedNodeIds(http, batch, log) + if (confirmation === null) { for (const id of batch) uncheckedNodeIds.add(id) return } - for (const id of edited) editedNodeIds.add(id) + for (const id of confirmation.editedIds) editedNodeIds.add(id) + for (const id of confirmation.unresolvedIds) uncheckedNodeIds.add(id) }) const isConfirmedEdited = (m: IShadowDiffMismatch) =>