houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Update fields select to People Data Labs enrichment functions (#21801)

Browse Source 
## What

The People Data Labs enrichment logic functions (`enrich-person`,
`enrich-company`, `enrich-people`, `enrich-companies`) now expose an
**`Update fields`** select instead of the `overrideExistingValues`
boolean, and always return the enriched data in their output.

`Update fields` options:
- **Yes and overwrite**: persist, overwriting existing standard fields
- **Yes and don't overwrite** (default): persist, filling standard
fields only when empty
- **No**: write nothing to the record (no CRM fields, no PDL metadata,
no company creation)

## Why

The functions previously only persisted data. With `No`, they can now
fetch from PDL and return the result without modifying the record, so
downstream workflow steps can consume it. Every matched result now
carries a `data` object with the mapped record fields (standard + `pdl*`
values), and the bulk functions also declare their `results[]` array in
the output schema.

Billing is unchanged: a successful PDL match is still charged in all
modes, since the API cost is incurred regardless of persistence.

## Notes

- Default behavior is preserved (unset input means fill-empty +
persist).
- Typecheck, lint, and the full unit suite (368 tests) pass.
This commit is contained in:
Raphaël Bosi
2026-06-19 14:39:49 +02:00
committed by GitHub
parent d19b7f8485
commit 88967a6e47
26 changed files with 570 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-apps/internal/people-data-labs/README.md
+10 -4
View File
@@ -16,23 +16,29 @@ Enriches **Person** and **Company** records with [People Data Labs](https://www.
`enrich-company` / `enrich-person` (single-record functions exposed **both** as a workflow action
and as an AI tool) all delegate to a shared, trigger-agnostic core in `src/logic-functions/handlers/`:
- The bulk workflow-action functions accept a **list of records** (`{ records, overrideExistingValues? }`),
- The bulk workflow-action functions accept a **list of records** (`{ records, updateFields? }`),
call the PDL **bulk** Enrichment endpoints (`/person/bulk`, `/company/enrich/bulk`), and loop the
single-record core over each, aggregating the outcome (`total` / `matched` / `notFound` / `skipped` /
`errored`); a per-record failure is captured as `ERROR` without aborting the batch
(`src/logic-functions/utils/run-batch-enrichment.ts`).
- The single-record functions accept one record (`{ recordId, overrideExistingValues? }`), call the PDL
- The single-record functions accept one record (`{ recordId, updateFields? }`), call the PDL
**single-record** Enrichment endpoints (`/person/enrich`, `/company/enrich` —
`src/logic-functions/utils/post-pdl-single-enrich.ts`), and return a single `EnrichResult`
(`src/logic-functions/utils/run-single-enrichment.ts`). They declare both a
`workflowActionTriggerSettings` and a `toolTriggerSettings`, so one function is usable as a workflow
step and as an AI tool.
- The **`updateFields`** select controls persistence: `Yes and overwrite` writes every enriched
standard field (replacing existing values), `Yes and don't overwrite` (the default) fills
standard fields only when empty, and `No` writes nothing to the record. In every mode each
matched result carries the enriched **mapped fields** under `data` (standard + `pdl*` values),
so `No` returns the data for downstream steps without modifying the record.
- Read the record, guard against re-enriching within a TTL (`pdlLastEnrichedAt`), pick a
match identifier (person: `pdlId` → LinkedIn → email → name; company: `pdlId` → domain →
name), and call the PDL Person/Company Enrichment API (`src/logic-functions/utils/`).
- On a match: fill **standard fields only when empty** (never clobber user data), always
(re)write `pdl*` fields, and set `pdlEnrichmentStatus = MATCHED`, `pdlLastEnrichedAt`,
- On a match: fill **standard fields** per `updateFields` (default: only when empty, never
clobbering user data); when `updateFields` is not `No`, (re)write `pdl*` fields and set
`pdlEnrichmentStatus = MATCHED`, `pdlLastEnrichedAt`,
`pdlRawPayload` (+ `pdlLikelihood` for Person). PDL `404` → `NOT_FOUND`; other errors →
`ERROR`. No identifier / fresh TTL → skipped with no writes.
- SELECT/MULTI_SELECT values are normalized and dropped if not in the field's option set
packages/twenty-apps/internal/people-data-labs/src/constants/update-fields-option-values.ts
+5
View File
@@ -0,0 +1,5 @@
import { UPDATE_FIELDS_OPTIONS } from 'src/constants/update-fields-options';
import { type UpdateFieldsOption } from 'src/types/update-fields-option';
export const UPDATE_FIELDS_OPTION_VALUES: UpdateFieldsOption[] =
Object.values(UPDATE_FIELDS_OPTIONS);
packages/twenty-apps/internal/people-data-labs/src/constants/update-fields-options.ts
+7
View File
@@ -0,0 +1,7 @@
import { type UpdateFieldsOption } from 'src/types/update-fields-option';
export const UPDATE_FIELDS_OPTIONS = {
overwrite: 'Yes and overwrite',
fillEmpty: "Yes and don't overwrite",
no: 'No',
} as const satisfies Record<string, UpdateFieldsOption>;
packages/twenty-apps/internal/people-data-labs/src/logic-functions/enrich-companies.function.ts
+24 -3
View File
@@ -1,5 +1,6 @@
import { defineLogicFunction } from 'twenty-sdk/define';
import { UPDATE_FIELDS_OPTION_VALUES } from 'src/constants/update-fields-option-values';
import { PDL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS } from 'src/constants/universal-identifiers';
import { enrichCompaniesCore } from 'src/logic-functions/handlers/enrich-companies';
import { type BulkEnrichInput } from 'src/types/bulk-enrich-input';
@@ -24,9 +25,10 @@ export default defineLogicFunction({
items: { type: 'object' },
label: 'Records',
},
overrideExistingValues: {
type: 'boolean',
label: 'Override Existing Values',
updateFields: {
type: 'string',
label: 'Update fields',
enum: [...UPDATE_FIELDS_OPTION_VALUES],
},
},
},
@@ -41,6 +43,25 @@ export default defineLogicFunction({
notFound: { type: 'number', label: 'Not Found' },
skipped: { type: 'number', label: 'Skipped' },
errored: { type: 'number', label: 'Errored' },
results: {
type: 'array',
items: {
type: 'object',
properties: {
success: { type: 'boolean', label: 'Success' },
recordId: { type: 'string', label: 'Record Id' },
status: { type: 'string', label: 'Status' },
updatedFields: {
type: 'array',
items: { type: 'string' },
label: 'Updated Fields',
},
data: { type: 'object', label: 'Data' },
message: { type: 'string', label: 'Message' },
},
},
label: 'Results',
},
},
},
],
packages/twenty-apps/internal/people-data-labs/src/logic-functions/enrich-company.function.ts
+10 -6
View File
@@ -1,5 +1,6 @@
import { defineLogicFunction } from 'twenty-sdk/define';
import { UPDATE_FIELDS_OPTION_VALUES } from 'src/constants/update-fields-option-values';
import { PDL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS } from 'src/constants/universal-identifiers';
import { enrichCompanyCore } from 'src/logic-functions/handlers/enrich-company';
import { type SingleEnrichInput } from 'src/types/single-enrich-input';
@@ -24,9 +25,10 @@ export default defineLogicFunction({
type: 'string',
label: 'Record',
},
overrideExistingValues: {
type: 'boolean',
label: 'Override Existing Values',
updateFields: {
type: 'string',
label: 'Update fields',
enum: [...UPDATE_FIELDS_OPTION_VALUES],
},
},
},
@@ -43,6 +45,7 @@ export default defineLogicFunction({
items: { type: 'string' },
label: 'Updated Fields',
},
data: { type: 'object', label: 'Data' },
message: { type: 'string', label: 'Message' },
},
},
@@ -56,10 +59,11 @@ export default defineLogicFunction({
type: 'string',
description: 'The id of the Company record to enrich.',
},
overrideExistingValues: {
type: 'boolean',
updateFields: {
type: 'string',
enum: [...UPDATE_FIELDS_OPTION_VALUES],
description:
'Overwrite existing field values with the enriched data instead of only filling empty fields.',
'Whether to write the enriched data back to the record. "Yes and overwrite" replaces existing values; "Yes and don\'t overwrite" only fills empty fields; "No" returns the enriched data without modifying the record.',
},
},
required: ['recordId'],
packages/twenty-apps/internal/people-data-labs/src/logic-functions/enrich-people.function.ts
+24 -3
View File
@@ -1,5 +1,6 @@
import { defineLogicFunction } from 'twenty-sdk/define';
import { UPDATE_FIELDS_OPTION_VALUES } from 'src/constants/update-fields-option-values';
import { PDL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS } from 'src/constants/universal-identifiers';
import { enrichPeopleCore } from 'src/logic-functions/handlers/enrich-people';
import { type BulkEnrichInput } from 'src/types/bulk-enrich-input';
@@ -24,9 +25,10 @@ export default defineLogicFunction({
items: { type: 'object' },
label: 'Records',
},
overrideExistingValues: {
type: 'boolean',
label: 'Override Existing Values',
updateFields: {
type: 'string',
label: 'Update fields',
enum: [...UPDATE_FIELDS_OPTION_VALUES],
},
},
},
@@ -41,6 +43,25 @@ export default defineLogicFunction({
notFound: { type: 'number', label: 'Not Found' },
skipped: { type: 'number', label: 'Skipped' },
errored: { type: 'number', label: 'Errored' },
results: {
type: 'array',
items: {
type: 'object',
properties: {
success: { type: 'boolean', label: 'Success' },
recordId: { type: 'string', label: 'Record Id' },
status: { type: 'string', label: 'Status' },
updatedFields: {
type: 'array',
items: { type: 'string' },
label: 'Updated Fields',
},
data: { type: 'object', label: 'Data' },
message: { type: 'string', label: 'Message' },
},
},
label: 'Results',
},
},
},
],
packages/twenty-apps/internal/people-data-labs/src/logic-functions/enrich-person.function.ts
+10 -6
View File
@@ -1,5 +1,6 @@
import { defineLogicFunction } from 'twenty-sdk/define';
import { UPDATE_FIELDS_OPTION_VALUES } from 'src/constants/update-fields-option-values';
import { PDL_LOGIC_FUNCTION_UNIVERSAL_IDENTIFIERS } from 'src/constants/universal-identifiers';
import { enrichPersonCore } from 'src/logic-functions/handlers/enrich-person';
import { type SingleEnrichInput } from 'src/types/single-enrich-input';
@@ -24,9 +25,10 @@ export default defineLogicFunction({
type: 'string',
label: 'Record',
},
overrideExistingValues: {
type: 'boolean',
label: 'Override Existing Values',
updateFields: {
type: 'string',
label: 'Update fields',
enum: [...UPDATE_FIELDS_OPTION_VALUES],
},
},
},
@@ -43,6 +45,7 @@ export default defineLogicFunction({
items: { type: 'string' },
label: 'Updated Fields',
},
data: { type: 'object', label: 'Data' },
message: { type: 'string', label: 'Message' },
},
},
@@ -56,10 +59,11 @@ export default defineLogicFunction({
type: 'string',
description: 'The id of the Person record to enrich.',
},
overrideExistingValues: {
type: 'boolean',
updateFields: {
type: 'string',
enum: [...UPDATE_FIELDS_OPTION_VALUES],
description:
'Overwrite existing field values with the enriched data instead of only filling empty fields.',
'Whether to write the enriched data back to the record. "Yes and overwrite" replaces existing values; "Yes and don\'t overwrite" only fills empty fields; "No" returns the enriched data without modifying the record.',
},
},
required: ['recordId'],
packages/twenty-apps/internal/people-data-labs/src/logic-functions/handlers/__tests__/enrich-people.spec.ts
+31 -2
View File
@@ -1,6 +1,7 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { type CoreApiClient } from 'twenty-client-sdk/core';
import { UPDATE_FIELDS_OPTIONS } from 'src/constants/update-fields-options';
import { createCoreApiClientMock } from 'src/logic-functions/__mocks__/create-core-api-client-mock';
import { PERSON_NODE_MOCK } from 'src/logic-functions/__mocks__/person-node.mock';
import { enrichPeopleCore } from 'src/logic-functions/handlers/enrich-people';
@@ -214,7 +215,7 @@ describe('enrichPeopleCore', () => {
});
});
it('overwrites a populated standard field when overrideExistingValues is set', async () => {
it('overwrites a populated standard field when updateFields is "Yes and overwrite"', async () => {
enrichPeopleMock.mockResolvedValue([
{
outcome: 'matched',
@@ -230,13 +231,41 @@ describe('enrichPeopleCore', () => {
});
await enrichPeopleCore({
input: { records: [{ id: 'p1' }], overrideExistingValues: true },
input: {
records: [{ id: 'p1' }],
updateFields: UPDATE_FIELDS_OPTIONS.overwrite,
},
client,
});
expect(captured.updatePerson?.jobTitle).toBe('CEO');
});
it('returns enriched data without writing when updateFields is "No"', async () => {
enrichPeopleMock.mockResolvedValue([
{
outcome: 'matched',
httpStatus: 200,
likelihood: 8,
data: { id: 'pdl1', first_name: 'Jane', job_title: 'CEO' },
},
]);
const captured: Captured = {};
const client = buildClient({ people: [PERSON_NODE_MOCK], captured });
const result = await enrichPeopleCore({
input: { records: [{ id: 'p1' }], updateFields: UPDATE_FIELDS_OPTIONS.no },
client,
});
expect(result.matched).toBe(1);
expect(result.results[0].updatedFields).toEqual([]);
expect(result.results[0].data?.jobTitle).toBe('CEO');
expect(captured.updatePerson).toBeUndefined();
expect(captured.updatePeople).toBeUndefined();
expect(captured.createCompanyCalled).toBeUndefined();
});
it('skips when there is no usable identifier', async () => {
const captured: Captured = {};
const client = buildClient({
packages/twenty-apps/internal/people-data-labs/src/logic-functions/handlers/__tests__/enrich-person.spec.ts
+27 -2
View File
@@ -1,6 +1,7 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { type CoreApiClient } from 'twenty-client-sdk/core';
import { UPDATE_FIELDS_OPTIONS } from 'src/constants/update-fields-options';
import { createCoreApiClientMock } from 'src/logic-functions/__mocks__/create-core-api-client-mock';
import { PERSON_NODE_MOCK } from 'src/logic-functions/__mocks__/person-node.mock';
import { enrichPersonCore } from 'src/logic-functions/handlers/enrich-person';
@@ -113,7 +114,7 @@ describe('enrichPersonCore', () => {
});
});
it('overwrites a populated standard field when overrideExistingValues is set', async () => {
it('overwrites a populated standard field when updateFields is "Yes and overwrite"', async () => {
enrichPersonMock.mockResolvedValue([
{
outcome: 'matched',
@@ -129,13 +130,37 @@ describe('enrichPersonCore', () => {
);
await enrichPersonCore({
input: { recordId: 'p1', overrideExistingValues: true },
input: { recordId: 'p1', updateFields: UPDATE_FIELDS_OPTIONS.overwrite },
client,
});
expect(captured.updatePerson?.jobTitle).toBe('CEO');
});
it('returns enriched data without writing when updateFields is "No"', async () => {
enrichPersonMock.mockResolvedValue([
{
outcome: 'matched',
httpStatus: 200,
likelihood: 8,
data: { id: 'pdl1', first_name: 'Jane', job_title: 'CEO' },
},
]);
const captured: Captured = {};
const client = buildClient([PERSON_NODE_MOCK], captured);
const result = await enrichPersonCore({
input: { recordId: 'p1', updateFields: UPDATE_FIELDS_OPTIONS.no },
client,
});
expect(result.status).toBe('MATCHED');
expect(result.updatedFields).toEqual([]);
expect(result.data?.jobTitle).toBe('CEO');
expect(captured.updatePerson).toBeUndefined();
expect(captured.updatePeople).toBeUndefined();
});
it('skips when there is no usable identifier', async () => {
const captured: Captured = {};
const client = buildClient(
packages/twenty-apps/internal/people-data-labs/src/logic-functions/handlers/company-enrichment-adapter.ts
+8 -1
View File
@@ -23,12 +23,19 @@ export const companyEnrichmentAdapter: BatchEnrichmentAdapter<
getNodeId: (node) => node.id,
extractParams: extractCompanyMatchParams,
enrichBatch: enrichCompanies,
buildMatchedData: ({ node, outcome, enrichedAt, overrideExistingValues }) =>
buildMatchedData: ({
node,
outcome,
enrichedAt,
overrideExistingValues,
shouldPersist,
}) =>
buildCompanyMatchedData({
node,
outcome,
enrichedAt,
overrideExistingValues,
shouldPersist,
}),
updateOne: updateCompanyRecord,
updateManyStatus: updateCompaniesStatus,
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/__tests__/build-company-matched-data.spec.ts
+29 -8
View File
@@ -8,28 +8,49 @@ const ENRICHED_AT = '2026-01-01T00:00:00.000Z';
describe('buildCompanyMatchedData', () => {
it('fills empty standard fields and always writes pdl metadata', async () => {
const data = await buildCompanyMatchedData({
const { mappedData, persistData } = await buildCompanyMatchedData({
node: COMPANY_NODE_MOCK,
outcome: { data: PDL_COMPANY_DATA_MOCK },
enrichedAt: ENRICHED_AT,
overrideExistingValues: false,
shouldPersist: true,
});
expect(data.name).toBe('Acme Corp');
expect(data.pdlIndustry).toBe('ACCOUNTING');
expect(data.pdlEnrichmentStatus).toBe('MATCHED');
expect(data.pdlLastEnrichedAt).toBe(ENRICHED_AT);
expect('domainName' in data).toBe(false);
expect(persistData.name).toBe('Acme Corp');
expect(persistData.pdlIndustry).toBe('ACCOUNTING');
expect(persistData.pdlEnrichmentStatus).toBe('MATCHED');
expect(persistData.pdlLastEnrichedAt).toBe(ENRICHED_AT);
expect('domainName' in persistData).toBe(false);
expect(mappedData.name).toBe('Acme Corp');
expect(mappedData.pdlIndustry).toBe('ACCOUNTING');
expect('pdlEnrichmentStatus' in mappedData).toBe(false);
expect('pdlRawPayload' in mappedData).toBe(false);
});
it('overwrites a populated standard field when overrideExistingValues is set', async () => {
const data = await buildCompanyMatchedData({
const { persistData } = await buildCompanyMatchedData({
node: COMPANY_NODE_MOCK,
outcome: { data: PDL_COMPANY_DATA_MOCK },
enrichedAt: ENRICHED_AT,
overrideExistingValues: true,
shouldPersist: true,
});
expect('domainName' in data).toBe(true);
expect('domainName' in persistData).toBe(true);
});
it('returns mapped data with no persist data when not persisting', async () => {
const { mappedData, persistData } = await buildCompanyMatchedData({
node: COMPANY_NODE_MOCK,
outcome: { data: PDL_COMPANY_DATA_MOCK },
enrichedAt: ENRICHED_AT,
overrideExistingValues: false,
shouldPersist: false,
});
expect(mappedData.name).toBe('Acme Corp');
expect(mappedData.pdlIndustry).toBe('ACCOUNTING');
expect(persistData).toEqual({});
});
});
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/__tests__/build-matched-result.spec.ts
+16
View File
@@ -11,7 +11,23 @@ describe('buildMatchedResult', () => {
recordId: 'p1',
status: 'MATCHED',
updatedFields: ['name', 'jobTitle'],
data: undefined,
message: 'Enriched with People Data Labs (2 fields).',
});
});
it('carries the enriched data when provided', () => {
expect(
buildMatchedResult({
recordId: 'p1',
updatedFields: [],
data: { name: { firstName: 'Jane' } },
}),
).toMatchObject({
status: 'MATCHED',
updatedFields: [],
data: { name: { firstName: 'Jane' } },
message: 'Matched People Data Labs data; no fields updated.',
});
});
});
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/__tests__/build-person-matched-data.spec.ts
+49 -11
View File
@@ -10,7 +10,7 @@ describe('buildPersonMatchedData', () => {
it('fills empty fields, writes pdl metadata, and skips company lookup when there is none', async () => {
const client = createCoreApiClientMock();
const data = await buildPersonMatchedData({
const { mappedData, persistData } = await buildPersonMatchedData({
client,
node: PERSON_NODE_MOCK,
outcome: {
@@ -26,29 +26,36 @@ describe('buildPersonMatchedData', () => {
enrichedAt: ENRICHED_AT,
companyIdByMatchKeyCache: new Map(),
overrideExistingValues: false,
shouldPersist: true,
});
expect(data.name).toEqual({ firstName: 'Jane', lastName: 'Doe' });
expect(data.jobTitle).toBe('CEO');
expect(data.pdlLikelihood).toBe(8);
expect(data.pdlEnrichmentStatus).toBe('MATCHED');
expect(data.pdlLastEnrichedAt).toBe(ENRICHED_AT);
expect('companyId' in data).toBe(false);
expect(persistData.name).toEqual({ firstName: 'Jane', lastName: 'Doe' });
expect(persistData.jobTitle).toBe('CEO');
expect(persistData.pdlLikelihood).toBe(8);
expect(persistData.pdlEnrichmentStatus).toBe('MATCHED');
expect(persistData.pdlLastEnrichedAt).toBe(ENRICHED_AT);
expect('companyId' in persistData).toBe(false);
expect(mappedData.name).toEqual({ firstName: 'Jane', lastName: 'Doe' });
expect(mappedData.jobTitle).toBe('CEO');
expect('pdlEnrichmentStatus' in mappedData).toBe(false);
expect('pdlRawPayload' in mappedData).toBe(false);
});
it('overwrites a populated standard field when overrideExistingValues is set', async () => {
const client = createCoreApiClientMock();
const data = await buildPersonMatchedData({
const { persistData } = await buildPersonMatchedData({
client,
node: { ...PERSON_NODE_MOCK, jobTitle: 'Existing Title' },
outcome: { likelihood: 8, data: { id: 'pdl1', job_title: 'CEO' } },
enrichedAt: ENRICHED_AT,
companyIdByMatchKeyCache: new Map(),
overrideExistingValues: true,
shouldPersist: true,
});
expect(data.jobTitle).toBe('CEO');
expect(persistData.jobTitle).toBe('CEO');
});
it('links a found-or-created company when the person has none', async () => {
@@ -57,7 +64,7 @@ describe('buildPersonMatchedData', () => {
mutationResult: { createCompany: { id: 'co-new' } },
});
const data = await buildPersonMatchedData({
const { persistData } = await buildPersonMatchedData({
client,
node: PERSON_NODE_MOCK,
outcome: {
@@ -72,8 +79,39 @@ describe('buildPersonMatchedData', () => {
enrichedAt: ENRICHED_AT,
companyIdByMatchKeyCache: new Map(),
overrideExistingValues: false,
shouldPersist: true,
});
expect(data.companyId).toBe('co-new');
expect(persistData.companyId).toBe('co-new');
});
it('returns mapped data with no persist data or company lookup when not persisting', async () => {
const client = createCoreApiClientMock({
queryResult: { companies: { edges: [] } },
mutationResult: { createCompany: { id: 'co-new' } },
});
const { mappedData, persistData } = await buildPersonMatchedData({
client,
node: PERSON_NODE_MOCK,
outcome: {
likelihood: 8,
data: {
id: 'pdl1',
first_name: 'Jane',
job_title: 'CEO',
job_company_name: 'Acme',
job_company_website: 'acme.com',
},
},
enrichedAt: ENRICHED_AT,
companyIdByMatchKeyCache: new Map(),
overrideExistingValues: false,
shouldPersist: false,
});
expect(mappedData.jobTitle).toBe('CEO');
expect(persistData).toEqual({});
expect(client.mutation).not.toHaveBeenCalled();
});
});
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/__tests__/resolve-update-fields-mode.spec.ts
+34
View File
@@ -0,0 +1,34 @@
import { describe, expect, it } from 'vitest';
import { UPDATE_FIELDS_OPTIONS } from 'src/constants/update-fields-options';
import { resolveUpdateFieldsMode } from 'src/logic-functions/utils/resolve-update-fields-mode';
describe('resolveUpdateFieldsMode', () => {
it('persists and overwrites for "Yes and overwrite"', () => {
expect(resolveUpdateFieldsMode(UPDATE_FIELDS_OPTIONS.overwrite)).toEqual({
shouldPersist: true,
overrideExistingValues: true,
});
});
it('persists without overwriting for "Yes and don\'t overwrite"', () => {
expect(resolveUpdateFieldsMode(UPDATE_FIELDS_OPTIONS.fillEmpty)).toEqual({
shouldPersist: true,
overrideExistingValues: false,
});
});
it('does not persist for "No"', () => {
expect(resolveUpdateFieldsMode(UPDATE_FIELDS_OPTIONS.no)).toEqual({
shouldPersist: false,
overrideExistingValues: false,
});
});
it('defaults to fill-empty persisting when omitted', () => {
expect(resolveUpdateFieldsMode(undefined)).toEqual({
shouldPersist: true,
overrideExistingValues: false,
});
});
});
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/__tests__/run-batch-enrichment.spec.ts
+41 -7
View File
@@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
import { type CoreApiClient } from 'twenty-client-sdk/core';
import { chargeCredits } from 'twenty-sdk/billing';
import { UPDATE_FIELDS_OPTIONS } from 'src/constants/update-fields-options';
import { runBatchEnrichment } from 'src/logic-functions/utils/run-batch-enrichment';
import { type BatchEnrichmentAdapter } from 'src/types/batch-enrichment-adapter';
import { type PdlEnrichResult } from 'src/types/pdl-enrich-result';
@@ -86,12 +87,17 @@ const buildHarness = (configs: RecordConfig[]) => {
const updateManyStatus = vi.fn(async () => undefined);
const buildMatchedData = vi.fn(
async ({ node }: BuildMatchedDataArgs): Promise<Record<string, unknown>> => {
async ({
node,
}: BuildMatchedDataArgs): Promise<{
mappedData: Record<string, unknown>;
persistData: Record<string, unknown>;
}> => {
if (byId.get(node.id)?.buildFails === true) {
throw new Error('build failed');
}
return { value: node.id };
return { mappedData: { mapped: node.id }, persistData: { value: node.id } };
},
);
@@ -203,21 +209,21 @@ describe('runBatchEnrichment', () => {
expect(result).toMatchObject({ skipped: 1, matched: 1 });
});
it('passes overrideExistingValues through to buildMatchedData', async () => {
it('maps "Yes and overwrite" to overrideExistingValues and persisting', async () => {
const harness = buildHarness([{ id: 'a' }]);
await runBatchEnrichment({
client: CLIENT,
input: { records: records('a'), overrideExistingValues: true },
input: { records: records('a'), updateFields: UPDATE_FIELDS_OPTIONS.overwrite },
adapter: harness.adapter,
});
expect(harness.buildMatchedData).toHaveBeenCalledWith(
expect.objectContaining({ overrideExistingValues: true }),
expect.objectContaining({ overrideExistingValues: true, shouldPersist: true }),
);
});
it('defaults overrideExistingValues to false when omitted', async () => {
it('defaults to fill-empty persisting when updateFields is omitted', async () => {
const harness = buildHarness([{ id: 'a' }]);
await runBatchEnrichment({
@@ -227,10 +233,38 @@ describe('runBatchEnrichment', () => {
});
expect(harness.buildMatchedData).toHaveBeenCalledWith(
expect.objectContaining({ overrideExistingValues: false }),
expect.objectContaining({
overrideExistingValues: false,
shouldPersist: true,
}),
);
});
it('returns mapped data without writing anything when updateFields is "No"', async () => {
const harness = buildHarness([
{ id: 'a' },
{ id: 'b', outcome: { outcome: 'not_found', httpStatus: 404 } },
]);
const result = await runBatchEnrichment({
client: CLIENT,
input: { records: records('a', 'b'), updateFields: UPDATE_FIELDS_OPTIONS.no },
adapter: harness.adapter,
});
expect(harness.buildMatchedData).toHaveBeenCalledWith(
expect.objectContaining({ shouldPersist: false }),
);
expect(harness.updateOne).not.toHaveBeenCalled();
expect(harness.updateManyStatus).not.toHaveBeenCalled();
const matched = result.results.find((entry) => entry.recordId === 'a');
expect(matched?.status).toBe('MATCHED');
expect(matched?.updatedFields).toEqual([]);
expect(matched?.data).toEqual({ mapped: 'a' });
expect(result).toMatchObject({ matched: 1, notFound: 1 });
});
it('marks missing records as ERROR without enriching them', async () => {
const harness = buildHarness([{ id: 'a' }, { id: 'b', exists: false }]);
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/build-company-matched-data.ts
+15 -2
View File
@@ -19,13 +19,24 @@ export const buildCompanyMatchedData = async ({
outcome,
enrichedAt,
overrideExistingValues,
shouldPersist,
}: {
node: CompanyNode;
outcome: { data: PdlCompanyData };
enrichedAt: string;
overrideExistingValues: boolean;
}): Promise<Record<string, unknown>> => {
shouldPersist: boolean;
}): Promise<{
mappedData: Record<string, unknown>;
persistData: Record<string, unknown>;
}> => {
const mapped = mapCompany(outcome.data);
const mappedData = pruneUndefined({ ...mapped.standard, ...mapped.pdl });
if (!shouldPersist) {
return { mappedData, persistData: {} };
}
const writableStandard = pickWritableStandard({
standard: mapped.standard,
current: node as unknown as Record<string, unknown>,
@@ -33,11 +44,13 @@ export const buildCompanyMatchedData = async ({
overrideExistingValues,
});
return pruneUndefined({
const persistData = pruneUndefined({
...writableStandard,
...mapped.pdl,
pdlRawPayload: outcome.data,
pdlLastEnrichedAt: enrichedAt,
pdlEnrichmentStatus: 'MATCHED',
});
return { mappedData, persistData };
};
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/build-matched-result.ts
+7 -1
View File
@@ -3,13 +3,19 @@ import { type EnrichResult } from 'src/types/enrich-result';
export const buildMatchedResult = ({
recordId,
updatedFields,
data,
}: {
recordId: string;
updatedFields: string[];
data?: Record<string, unknown>;
}): EnrichResult => ({
success: true,
recordId,
status: 'MATCHED',
updatedFields,
message: `Enriched with People Data Labs (${updatedFields.length} fields).`,
data,
message:
updatedFields.length > 0
? `Enriched with People Data Labs (${updatedFields.length} fields).`
: 'Matched People Data Labs data; no fields updated.',
});
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/build-person-matched-data.ts
+15 -2
View File
@@ -29,6 +29,7 @@ export const buildPersonMatchedData = async ({
enrichedAt,
companyIdByMatchKeyCache,
overrideExistingValues,
shouldPersist,
}: {
client: CoreApiClient;
node: PersonNode;
@@ -36,8 +37,18 @@ export const buildPersonMatchedData = async ({
enrichedAt: string;
companyIdByMatchKeyCache: CompanyIdByMatchKeyCache;
overrideExistingValues: boolean;
}): Promise<Record<string, unknown>> => {
shouldPersist: boolean;
}): Promise<{
mappedData: Record<string, unknown>;
persistData: Record<string, unknown>;
}> => {
const mapped = mapPerson(outcome.data);
const mappedData = pruneUndefined({ ...mapped.standard, ...mapped.pdl });
if (!shouldPersist) {
return { mappedData, persistData: {} };
}
const writableStandard = pickWritableStandard({
standard: mapped.standard,
current: node as unknown as Record<string, unknown>,
@@ -53,7 +64,7 @@ export const buildPersonMatchedData = async ({
companyIdByMatchKeyCache,
});
return pruneUndefined({
const persistData = pruneUndefined({
...writableStandard,
...mapped.pdl,
companyId: currentCompanyId,
@@ -62,4 +73,6 @@ export const buildPersonMatchedData = async ({
pdlLastEnrichedAt: enrichedAt,
pdlEnrichmentStatus: 'MATCHED',
});
return { mappedData, persistData };
};
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/enrich-chunk.ts
+175 -82
View File
@@ -7,6 +7,7 @@ import { buildSkippedResult } from 'src/logic-functions/utils/build-skipped-resu
import { chargeMatchedEnrichments } from 'src/logic-functions/utils/charge-matched-enrichments';
import { INTERNAL_BOOKKEEPING_FIELDS } from 'src/logic-functions/utils/internal-field-names';
import { nowIso } from 'src/logic-functions/utils/now-iso';
import { resolveUpdateFieldsMode } from 'src/logic-functions/utils/resolve-update-fields-mode';
import { type BatchEnrichmentAdapter } from 'src/types/batch-enrichment-adapter';
import { type BulkEnrichInput } from 'src/types/bulk-enrich-input';
import { type CompanyIdByMatchKeyCache } from 'src/types/company-id-by-match-key-cache';
@@ -39,6 +40,137 @@ const writeErrorStatusWithBackoff = async <TNode, TData, TParams>({
.catch(() => undefined);
};
type MatchedRecord = {
recordId: string;
mappedData: Record<string, unknown>;
persistData: Record<string, unknown>;
};
const recordMatchedRecords = async <TNode, TData, TParams>({
adapter,
client,
matchedRecords,
shouldPersist,
resultById,
}: {
adapter: BatchEnrichmentAdapter<TNode, TData, TParams>;
client: CoreApiClient;
matchedRecords: MatchedRecord[];
shouldPersist: boolean;
resultById: Map<string, EnrichResult>;
}): Promise<string[]> => {
if (matchedRecords.length === 0) {
return [];
}
if (!shouldPersist) {
for (const matchedRecord of matchedRecords) {
resultById.set(
matchedRecord.recordId,
buildMatchedResult({
recordId: matchedRecord.recordId,
updatedFields: [],
data: matchedRecord.mappedData,
}),
);
}
return [];
}
const settledWriteResults = await Promise.allSettled(
matchedRecords.map((matchedRecord) =>
adapter.updateOne({
client,
recordId: matchedRecord.recordId,
data: matchedRecord.persistData,
}),
),
);
const failedRecordIds: string[] = [];
for (const [index, writeResult] of settledWriteResults.entries()) {
const matchedRecord = matchedRecords[index];
if (writeResult.status === 'rejected') {
resultById.set(
matchedRecord.recordId,
buildErrorResult({
recordId: matchedRecord.recordId,
error: toErrorMessage(writeResult.reason),
}),
);
failedRecordIds.push(matchedRecord.recordId);
continue;
}
resultById.set(
matchedRecord.recordId,
buildMatchedResult({
recordId: matchedRecord.recordId,
updatedFields: Object.keys(matchedRecord.persistData).filter(
(fieldName) => !INTERNAL_BOOKKEEPING_FIELDS.has(fieldName),
),
data: matchedRecord.mappedData,
}),
);
}
return failedRecordIds;
};
const recordNotFoundRecords = async <TNode, TData, TParams>({
adapter,
client,
notFoundRecordIds,
shouldPersist,
enrichedAt,
resultById,
}: {
adapter: BatchEnrichmentAdapter<TNode, TData, TParams>;
client: CoreApiClient;
notFoundRecordIds: string[];
shouldPersist: boolean;
enrichedAt: string;
resultById: Map<string, EnrichResult>;
}): Promise<string[]> => {
if (notFoundRecordIds.length === 0) {
return [];
}
if (!shouldPersist) {
for (const recordId of notFoundRecordIds) {
resultById.set(recordId, buildNotFoundResult(recordId));
}
return [];
}
try {
await adapter.updateManyStatus({
client,
recordIds: notFoundRecordIds,
data: { pdlEnrichmentStatus: 'NOT_FOUND', pdlLastEnrichedAt: enrichedAt },
});
for (const recordId of notFoundRecordIds) {
resultById.set(recordId, buildNotFoundResult(recordId));
}
return [];
} catch (notFoundStatusWriteError) {
const notFoundStatusWriteErrorMessage = toErrorMessage(
notFoundStatusWriteError,
);
for (const recordId of notFoundRecordIds) {
resultById.set(
recordId,
buildErrorResult({ recordId, error: notFoundStatusWriteErrorMessage }),
);
}
return notFoundRecordIds;
}
};
export const enrichChunk = async <TNode, TData, TParams>({
client,
recordIds,
@@ -54,6 +186,10 @@ export const enrichChunk = async <TNode, TData, TParams>({
resultById: Map<string, EnrichResult>;
companyIdByMatchKeyCache: CompanyIdByMatchKeyCache;
}): Promise<void> => {
const { shouldPersist, overrideExistingValues } = resolveUpdateFieldsMode(
input.updateFields,
);
let recordNodes: TNode[];
try {
recordNodes = await adapter.readRecords({ client, recordIds });
@@ -126,12 +262,16 @@ export const enrichChunk = async <TNode, TData, TParams>({
}),
);
}
await writeErrorStatusWithBackoff({
adapter,
client,
recordIds: recordsToEnrich.map((recordToEnrich) => recordToEnrich.recordId),
enrichedAt,
});
if (shouldPersist) {
await writeErrorStatusWithBackoff({
adapter,
client,
recordIds: recordsToEnrich.map(
(recordToEnrich) => recordToEnrich.recordId,
),
enrichedAt,
});
}
return;
}
@@ -145,10 +285,7 @@ export const enrichChunk = async <TNode, TData, TParams>({
});
const notFoundRecordIds: string[] = [];
const matchedRecordsToPersist: {
recordId: string;
data: Record<string, unknown>;
}[] = [];
const matchedRecords: MatchedRecord[] = [];
for (let index = 0; index < recordsToEnrich.length; index++) {
const { recordId, node: recordNode } = recordsToEnrich[index];
@@ -172,15 +309,16 @@ export const enrichChunk = async <TNode, TData, TParams>({
}
try {
const matchedRecordData = await adapter.buildMatchedData({
const { mappedData, persistData } = await adapter.buildMatchedData({
client,
node: recordNode,
outcome: enrichmentOutcome,
enrichedAt,
companyIdByMatchKeyCache,
overrideExistingValues: input.overrideExistingValues === true,
overrideExistingValues,
shouldPersist,
});
matchedRecordsToPersist.push({ recordId, data: matchedRecordData });
matchedRecords.push({ recordId, mappedData, persistData });
} catch (buildMatchedDataError) {
resultById.set(
recordId,
@@ -193,76 +331,31 @@ export const enrichChunk = async <TNode, TData, TParams>({
}
}
if (matchedRecordsToPersist.length > 0) {
const settledWriteResults = await Promise.allSettled(
matchedRecordsToPersist.map((matchedRecord) =>
adapter.updateOne({
client,
recordId: matchedRecord.recordId,
data: matchedRecord.data,
}),
),
);
settledWriteResults.forEach((writeResult, index) => {
const matchedRecord = matchedRecordsToPersist[index];
if (writeResult.status === 'fulfilled') {
resultById.set(
matchedRecord.recordId,
buildMatchedResult({
recordId: matchedRecord.recordId,
updatedFields: Object.keys(matchedRecord.data).filter(
(fieldName) => !INTERNAL_BOOKKEEPING_FIELDS.has(fieldName),
),
}),
);
} else {
resultById.set(
matchedRecord.recordId,
buildErrorResult({
recordId: matchedRecord.recordId,
error: toErrorMessage(writeResult.reason),
}),
);
recordIdsToMarkAsError.push(matchedRecord.recordId);
}
});
}
if (notFoundRecordIds.length > 0) {
try {
await adapter.updateManyStatus({
client,
recordIds: notFoundRecordIds,
data: {
pdlEnrichmentStatus: 'NOT_FOUND',
pdlLastEnrichedAt: enrichedAt,
},
});
for (const recordId of notFoundRecordIds) {
resultById.set(recordId, buildNotFoundResult(recordId));
}
} catch (notFoundStatusWriteError) {
const notFoundStatusWriteErrorMessage = toErrorMessage(
notFoundStatusWriteError,
);
for (const recordId of notFoundRecordIds) {
resultById.set(
recordId,
buildErrorResult({
recordId,
error: notFoundStatusWriteErrorMessage,
}),
);
recordIdsToMarkAsError.push(recordId);
}
}
}
await writeErrorStatusWithBackoff({
const failedMatchedWriteRecordIds = await recordMatchedRecords({
adapter,
client,
recordIds: recordIdsToMarkAsError,
enrichedAt,
matchedRecords,
shouldPersist,
resultById,
});
recordIdsToMarkAsError.push(...failedMatchedWriteRecordIds);
const failedNotFoundWriteRecordIds = await recordNotFoundRecords({
adapter,
client,
notFoundRecordIds,
shouldPersist,
enrichedAt,
resultById,
});
recordIdsToMarkAsError.push(...failedNotFoundWriteRecordIds);
if (shouldPersist) {
await writeErrorStatusWithBackoff({
adapter,
client,
recordIds: recordIdsToMarkAsError,
enrichedAt,
});
}
};
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/resolve-update-fields-mode.ts
+16
View File
@@ -0,0 +1,16 @@
import { UPDATE_FIELDS_OPTIONS } from 'src/constants/update-fields-options';
import { type UpdateFieldsOption } from 'src/types/update-fields-option';
export const resolveUpdateFieldsMode = (
updateFields?: UpdateFieldsOption,
): { shouldPersist: boolean; overrideExistingValues: boolean } => {
if (updateFields === UPDATE_FIELDS_OPTIONS.no) {
return { shouldPersist: false, overrideExistingValues: false };
}
if (updateFields === UPDATE_FIELDS_OPTIONS.overwrite) {
return { shouldPersist: true, overrideExistingValues: true };
}
return { shouldPersist: true, overrideExistingValues: false };
};
packages/twenty-apps/internal/people-data-labs/src/logic-functions/utils/run-single-enrichment.ts
+1 -1
View File
@@ -31,7 +31,7 @@ export const runSingleEnrichment = async <TNode, TData, TParams>({
client,
input: {
records: recordId,
overrideExistingValues: input.overrideExistingValues,
updateFields: input.updateFields,
minLikelihood: input.minLikelihood,
},
adapter,
packages/twenty-apps/internal/people-data-labs/src/types/batch-enrichment-adapter.ts
+5 -1
View File
@@ -25,7 +25,11 @@ export type BatchEnrichmentAdapter<TNode, TData, TParams> = {
enrichedAt: string;
companyIdByMatchKeyCache: CompanyIdByMatchKeyCache;
overrideExistingValues: boolean;
}) => Promise<Record<string, unknown>>;
shouldPersist: boolean;
}) => Promise<{
mappedData: Record<string, unknown>;
persistData: Record<string, unknown>;
}>;
updateOne: (args: {
client: CoreApiClient;
recordId: string;
packages/twenty-apps/internal/people-data-labs/src/types/bulk-enrich-input.ts
+3 -1
View File
@@ -1,7 +1,9 @@
import { type UpdateFieldsOption } from 'src/types/update-fields-option';
export type RecordInput = string | { id?: string | null };
export type BulkEnrichInput = {
records: RecordInput | RecordInput[];
overrideExistingValues?: boolean;
updateFields?: UpdateFieldsOption;
minLikelihood?: number;
};
packages/twenty-apps/internal/people-data-labs/src/types/enrich-result.ts
+1
View File
@@ -5,6 +5,7 @@ export type EnrichResult = {
recordId: string;
status: EnrichStatus;
updatedFields: string[];
data?: Record<string, unknown>;
message: string;
error?: string;
};
packages/twenty-apps/internal/people-data-labs/src/types/single-enrich-input.ts
+3 -1
View File
@@ -1,5 +1,7 @@
import { type UpdateFieldsOption } from 'src/types/update-fields-option';
export type SingleEnrichInput = {
recordId?: string;
overrideExistingValues?: boolean;
updateFields?: UpdateFieldsOption;
minLikelihood?: number;
};
packages/twenty-apps/internal/people-data-labs/src/types/update-fields-option.ts
+4
View File
@@ -0,0 +1,4 @@
export type UpdateFieldsOption =
| 'Yes and overwrite'
| "Yes and don't overwrite"
| 'No';