diff --git a/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts b/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts index fa8915419c..434fac681e 100644 --- a/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts +++ b/services/apps/connectors_worker/src/activities/shadowDiffActivities.ts @@ -16,6 +16,11 @@ import { hasDeletedRecordCandidates, isDeletedRecordCandidate, } from '../deletedRecords' +import { + dropConfirmedEditedRecords, + hasEditedRecordCandidates, + isEditedRecordCandidate, +} from '../editedRecords' import { dropConfirmedForcePushedCommits, hasForcePushCandidates, @@ -97,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, } } @@ -180,8 +186,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 { @@ -195,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 @@ -256,6 +264,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..abdec57a1b --- /dev/null +++ b/services/apps/connectors_worker/src/editedRecords.test.ts @@ -0,0 +1,175 @@ +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) + }) + + 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', () => { + 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 and counts them unconfirmed', 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(1) + }) + + 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..630e949c36 --- /dev/null +++ b/services/apps/connectors_worker/src/editedRecords.ts @@ -0,0 +1,144 @@ +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 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 +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 = PARENT_BODY_SYNTHETIC_SOURCE_ID.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 +} + +interface BatchConfirmation { + editedIds: Set + unresolvedIds: Set +} + +async function fetchEditedNodeIds( + http: ConnectorHttp, + ids: string[], + log: Logger, +): Promise { + 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 + } + 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 + } +} + +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 confirmation = await fetchEditedNodeIds(http, batch, log) + if (confirmation === null) { + for (const id of batch) uncheckedNodeIds.add(id) + return + } + for (const id of confirmation.editedIds) editedNodeIds.add(id) + for (const id of confirmation.unresolvedIds) uncheckedNodeIds.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, + } +}