|
1 | | -import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' |
| 1 | +import { QueryExecutor, formatQuery } from '@crowd/data-access-layer/src/queryExecutor' |
2 | 2 | import { prepareBulkInsert } from '@crowd/data-access-layer/src/utils' |
3 | 3 |
|
4 | | -import { RepoPolicies, ScoredContact } from './types' |
| 4 | +import { ProcessRepoResult, RepoPolicies, ScoredContact } from './types' |
| 5 | + |
| 6 | +const CONTACT_COLUMNS = [ |
| 7 | + 'repo_id', |
| 8 | + 'channel', |
| 9 | + 'value', |
| 10 | + 'role', |
| 11 | + 'name', |
| 12 | + 'score', |
| 13 | + 'confidence', |
| 14 | + 'provenance', |
| 15 | + 'reachable', |
| 16 | + 'reachability_reason', |
| 17 | +] |
| 18 | + |
| 19 | +const CONTACT_UPSERT_SET = `(repo_id, channel, value) DO UPDATE SET |
| 20 | + role = EXCLUDED.role, |
| 21 | + name = EXCLUDED.name, |
| 22 | + score = EXCLUDED.score, |
| 23 | + confidence = EXCLUDED.confidence, |
| 24 | + provenance = EXCLUDED.provenance, |
| 25 | + reachable = EXCLUDED.reachable, |
| 26 | + reachability_reason = EXCLUDED.reachability_reason, |
| 27 | + last_refreshed = NOW(), |
| 28 | + updated_at = NOW(), |
| 29 | + deleted_at = NULL` |
| 30 | + |
| 31 | +function toContactRow(repoId: string, c: ScoredContact) { |
| 32 | + return { |
| 33 | + repo_id: repoId, |
| 34 | + channel: c.channel, |
| 35 | + value: c.value, |
| 36 | + role: c.role, |
| 37 | + name: c.name ?? null, |
| 38 | + score: c.score, |
| 39 | + confidence: c.confidence, |
| 40 | + provenance: JSON.stringify(c.provenance), |
| 41 | + reachable: c.reachable, |
| 42 | + reachability_reason: c.reachabilityReason, |
| 43 | + } |
| 44 | +} |
5 | 45 |
|
6 | 46 | // Advances contacts_last_refreshed only, so a failed pass isn't reprocessed this sweep. |
7 | 47 | export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Promise<void> { |
@@ -30,41 +70,9 @@ export async function writeContacts( |
30 | 70 | await tx.result( |
31 | 71 | prepareBulkInsert( |
32 | 72 | 'security_contacts', |
33 | | - [ |
34 | | - 'repo_id', |
35 | | - 'channel', |
36 | | - 'value', |
37 | | - 'role', |
38 | | - 'name', |
39 | | - 'score', |
40 | | - 'confidence', |
41 | | - 'provenance', |
42 | | - 'reachable', |
43 | | - 'reachability_reason', |
44 | | - ], |
45 | | - contacts.map((c) => ({ |
46 | | - repo_id: repoId, |
47 | | - channel: c.channel, |
48 | | - value: c.value, |
49 | | - role: c.role, |
50 | | - name: c.name ?? null, |
51 | | - score: c.score, |
52 | | - confidence: c.confidence, |
53 | | - provenance: JSON.stringify(c.provenance), |
54 | | - reachable: c.reachable, |
55 | | - reachability_reason: c.reachabilityReason, |
56 | | - })), |
57 | | - `(repo_id, channel, value) DO UPDATE SET |
58 | | - role = EXCLUDED.role, |
59 | | - name = EXCLUDED.name, |
60 | | - score = EXCLUDED.score, |
61 | | - confidence = EXCLUDED.confidence, |
62 | | - provenance = EXCLUDED.provenance, |
63 | | - reachable = EXCLUDED.reachable, |
64 | | - reachability_reason = EXCLUDED.reachability_reason, |
65 | | - last_refreshed = NOW(), |
66 | | - updated_at = NOW(), |
67 | | - deleted_at = NULL`, |
| 73 | + CONTACT_COLUMNS, |
| 74 | + contacts.map((c) => toContactRow(repoId, c)), |
| 75 | + CONTACT_UPSERT_SET, |
68 | 76 | ), |
69 | 77 | ) |
70 | 78 | } |
@@ -94,3 +102,90 @@ export async function writeContacts( |
94 | 102 | ) |
95 | 103 | }) |
96 | 104 | } |
| 105 | + |
| 106 | +export async function markReposAttempted(qx: QueryExecutor, repoIds: string[]): Promise<void> { |
| 107 | + if (repoIds.length === 0) return |
| 108 | + await qx.result( |
| 109 | + 'UPDATE repos SET contacts_last_refreshed = NOW() WHERE id = ANY($(repoIds)::bigint[])', |
| 110 | + { repoIds }, |
| 111 | + ) |
| 112 | +} |
| 113 | + |
| 114 | +type OkResult = Extract<ProcessRepoResult, { status: 'ok' }> |
| 115 | + |
| 116 | +// Bounds blast radius: a bad row only forces a re-extract of its chunk next sweep, not the whole batch. |
| 117 | +const WRITE_CHUNK_SIZE = 100 |
| 118 | + |
| 119 | +function prepareBulkPolicyUpdate(chunk: OkResult[]): string { |
| 120 | + const rows = chunk.map( |
| 121 | + (_, i) => |
| 122 | + `($(rows.id${i}), $(rows.securityPolicyUrl${i}), $(rows.vulnerabilityReportingUrl${i}), $(rows.pvrResolved${i}), $(rows.bugBountyUrl${i}), $(rows.securityTxtUrl${i}), $(rows.pvrEnabled${i}))`, |
| 123 | + ) |
| 124 | + |
| 125 | + const params = chunk.reduce( |
| 126 | + (acc, r, i) => { |
| 127 | + acc[`id${i}`] = r.repoId |
| 128 | + acc[`securityPolicyUrl${i}`] = r.policies.securityPolicyUrl ?? null |
| 129 | + acc[`vulnerabilityReportingUrl${i}`] = r.policies.vulnerabilityReportingUrl ?? null |
| 130 | + acc[`pvrResolved${i}`] = r.policies.pvrEnabled !== undefined |
| 131 | + acc[`bugBountyUrl${i}`] = r.policies.bugBountyUrl ?? null |
| 132 | + acc[`securityTxtUrl${i}`] = r.policies.securityTxtUrl ?? null |
| 133 | + acc[`pvrEnabled${i}`] = r.policies.pvrEnabled ?? null |
| 134 | + return acc |
| 135 | + }, |
| 136 | + {} as Record<string, unknown>, |
| 137 | + ) |
| 138 | + |
| 139 | + return formatQuery( |
| 140 | + `UPDATE repos AS r SET |
| 141 | + security_policy_url = COALESCE(v.security_policy_url, r.security_policy_url), |
| 142 | + vulnerability_reporting_url = CASE WHEN v.pvr_resolved |
| 143 | + THEN v.vulnerability_reporting_url |
| 144 | + ELSE COALESCE(v.vulnerability_reporting_url, r.vulnerability_reporting_url) |
| 145 | + END, |
| 146 | + bug_bounty_url = COALESCE(v.bug_bounty_url, r.bug_bounty_url), |
| 147 | + security_txt_url = COALESCE(v.security_txt_url, r.security_txt_url), |
| 148 | + pvr_enabled = COALESCE(v.pvr_enabled, r.pvr_enabled), |
| 149 | + contacts_last_refreshed = NOW() |
| 150 | + FROM (VALUES ${rows.join(',')}) AS v(id, security_policy_url, vulnerability_reporting_url, pvr_resolved, bug_bounty_url, security_txt_url, pvr_enabled) |
| 151 | + WHERE r.id = v.id::bigint`, |
| 152 | + params, |
| 153 | + ) |
| 154 | +} |
| 155 | + |
| 156 | +async function writeContactsChunk(qx: QueryExecutor, chunk: OkResult[]): Promise<void> { |
| 157 | + if (chunk.length === 0) return |
| 158 | + const repoIds = chunk.map((r) => r.repoId) |
| 159 | + |
| 160 | + await qx.tx(async (tx) => { |
| 161 | + await tx.result( |
| 162 | + 'UPDATE security_contacts SET deleted_at = NOW(), updated_at = NOW() WHERE repo_id = ANY($(repoIds)::bigint[]) AND deleted_at IS NULL', |
| 163 | + { repoIds }, |
| 164 | + ) |
| 165 | + |
| 166 | + const rows = chunk.flatMap((r) => r.contacts.map((c) => toContactRow(r.repoId, c))) |
| 167 | + if (rows.length > 0) { |
| 168 | + await tx.result( |
| 169 | + prepareBulkInsert('security_contacts', CONTACT_COLUMNS, rows, CONTACT_UPSERT_SET), |
| 170 | + ) |
| 171 | + } |
| 172 | + |
| 173 | + await tx.result(prepareBulkPolicyUpdate(chunk)) |
| 174 | + }) |
| 175 | +} |
| 176 | + |
| 177 | +export async function writeContactsBatch( |
| 178 | + qx: QueryExecutor, |
| 179 | + outcomes: ProcessRepoResult[], |
| 180 | +): Promise<void> { |
| 181 | + const attemptedOnlyIds = outcomes |
| 182 | + .filter((o) => o.status === 'extractor-failed') |
| 183 | + .map((o) => o.repoId) |
| 184 | + const ok = outcomes.filter((o): o is OkResult => o.status === 'ok') |
| 185 | + |
| 186 | + for (let i = 0; i < ok.length; i += WRITE_CHUNK_SIZE) { |
| 187 | + await writeContactsChunk(qx, ok.slice(i, i + WRITE_CHUNK_SIZE)) |
| 188 | + } |
| 189 | + |
| 190 | + await markReposAttempted(qx, attemptedOnlyIds) |
| 191 | +} |
0 commit comments