[HACKTOBERFEST] Add rollup engine app with UI-driven configuration (#15482)

## Summary - add packages/twenty-apps/rollup-engine: a parameterised rollup engine that ships a default Opportunity → Company aggregation. - declare runtime config in package.json (TWENTY_API_KEY, optional TWENTY_API_BASE_URL, and ROLLUP_ENGINE_CONFIG) so app configuration lives entirely in Settings → Apps → [App] → Configuration. - document the workflow in README.md: deploy via twenty app sync, populate env vars in the UI, use the Test panel with a ready JSON payload, and reference troubleshooting tips. - adjust the function entry point to a named async export and add fallback logic for blank base URLs, matching the UI’s env behaviour. - prune legacy templates and examples so config/templates/opportunity-to-company.json is the single copy/paste starting point. ## UI / UX impact After syncing the app: - the Configuration screen shows the three env keys with helpful descriptions (API key, optional base URL, JSON override), - the built-in Test your function panel works immediately - and the default JSON config is available from config/templates/opportunity-to-company.json for users who need to customise rollups. ## Testing - Out-of-the-box deploy to the hosted workspace (Opportunity update) ✓ - “Test your function” with the default config ✓ - Override example: point debugOpportunityCount at a scratch field via ROLLUP_ENGINE_CONFIG ✓ - Optional: local smoke test (yarn install && yarn smoke) still passes
2025-11-04 11:03:00 +00:00
parent fd5c39b4da
commit 7ec56433c7
15 changed files with 3762 additions and 0 deletions
@@ -0,0 +1,107 @@
+# Rollup engine
+
+General-purpose rollup engine for Twenty workspaces. The bundle ships a
+serverless function that materialises aggregations from a child object onto a
+parent object. The default configuration targets Opportunity → Company rollups.
+
+## Requirements
+- `twenty-cli` `npm install -g twenty-cli`
+- an API key with access to your workspace. Generate one at
+  `https://twenty.com/settings/api-webhooks`.
+- Node 20+ (only required if you plan to run the optional smoke test locally).
+
+## Metadata prerequisites
+- Opportunity is a standard object, no extra provisioning required.
+- Add these fields to the `company` object (API names shown):
+  - `totalPipelineAmount` (Currency)
+  - `totalOpportunityCount` (Number)
+  - `wonPipelineAmount` (Currency)
+  - `wonOpportunityCount` (Number)
+  - `openPipelineAmount` (Currency)
+  - `openOpportunityCount` (Number)
+  - `lastOpportunityCloseDate` (Date)
+- To script the setup, export `TWENTY_API_KEY` (and `TWENTY_METADATA_BASE_URL` if you are not targeting `http://localhost:3000/rest/metadata`) and run `yarn setup:metadata`. The helper script calls the Metadata API to ensure each field exists on the Company object, skipping anything already provisioned.
+
+Create the fields before syncing so PATCH requests succeed.
+
+## Quick start
+
+1. **Deploy the app**
+   ```bash
+   twenty auth login
+   cd rollup-engine
+   twenty app sync
+   ```
+
+2. **Configure environment variables**
+   - Open **Settings → Apps → Rollup engine → Configuration**.
+   - Provide values for the manifest-defined keys:
+     - `TWENTY_API_KEY` (required secret; create it in Twenty → Settings → API & Webhooks).
+     - `TWENTY_API_BASE_URL` (optional; leave blank to use `https://app.twenty.com/rest`).
+     - `ROLLUP_ENGINE_CONFIG` (optional JSON override; leave blank to use the baked-in config).
+   - Save the configuration. Changes are applied immediately—no redeploy required.
+
+3. **Verify via the Test panel**
+   - Still on the app page, select **Test**.
+   - Paste a JSON payload with the company ID you want to validate:
+
+     ```json
+     {
+       "trigger": { "type": "databaseEvent", "eventName": "opportunity.updated" },
+       "record": { "companyId": "YOUR_COMPANY_ID" },
+       "opportunity": { "companyId": "YOUR_COMPANY_ID" }
+     }
+     ```
+
+   - Click **Run**. You should see a success summary and the company’s rollup fields updating.
+
+4. **Trigger live rollups**
+   - Any Opportunity create/update/delete (or the nightly cron) now feeds into the Company metrics.
+
+## Runtime configuration
+- `serverlessFunctions/calculaterollups/src/rollupConfig.ts` holds the baked-in defaults.
+  Override them by setting `ROLLUP_ENGINE_CONFIG` (aliases: `ROLLUPS_CONFIG`,
+  `CALCULATE_ROLLUPS_CONFIG`) to a JSON array in the Configuration UI.
+- `config/templates/opportunity-to-company.json` matches the default rollups shipped with the bundle.
+  Copy it, tweak the object/field names you need, and paste the edited JSON into `ROLLUP_ENGINE_CONFIG`.
+- `filters[].dynamicValue` currently supports `"startOfYear"` (UTC midnight on the first day of the
+  current calendar year).
+
+### Default Opportunity → Company rollups
+| Parent field | Suggested type | Description |
+| --- | --- | --- |
+| `totalPipelineAmount` | Currency (`amountMicros`, `currencyCode`) | Sum of positive-valued Opportunities linked to the Company. |
+| `totalOpportunityCount` | Number | Count of Opportunities linked to the Company. |
+| `wonPipelineAmount` | Currency (`amountMicros`, `currencyCode`) | Sum of Opportunities where `stage === "CUSTOMER"`. |
+| `wonOpportunityCount` | Number | Count of Opportunities where `stage === "CUSTOMER"`. |
+| `openPipelineAmount` | Currency (`amountMicros`, `currencyCode`) | Sum of Opportunities where `stage !== "CUSTOMER"`. |
+| `openOpportunityCount` | Number | Count of Opportunities where `stage !== "CUSTOMER"`. |
+| `lastOpportunityCloseDate` | Date | Most recent `closeDate` (ISO date). |
+
+Assumptions:
+- Opportunities expose `companyId`, `amount.amountMicros`, `amount.currencyCode`,
+  `stage`, and `closeDate`. Adjust filters or fields as needed via the JSON
+  config.
+- Amount totals are stored in micros (integers) to align with Twenty’s composite
+  currency fields.
+
+## Optional local smoke test
+Run this if you want to exercise the aggregation logic without calling the live API.
+
+```bash
+cd rollup-engine
+yarn install
+yarn smoke
+```
+
+The smoke script replaces `fetch` with an in-memory mock, executes the rollup function using sample
+Opportunity data, and asserts the PATCH payload that would be sent to Twenty.
+
+## Troubleshooting
+- **`Test your function` reports `mainFile.main is not a function`:** delete the app
+  (`twenty app delete`) and sync it again (`twenty app sync`) to clear any stale bundle.
+- **`status: "error", message: "Unable to parse rollup configuration override"`:** confirm that
+  `ROLLUP_ENGINE_CONFIG` contains valid JSON or leave it blank to fall back to the baked-in defaults.
+- **No company updates after configuring everything:** double-check the Company field API names,
+  confirm the Opportunity records have `amount.amountMicros > 0`, and use the Test panel to inspect
+  the summary details/logs.
@@ -0,0 +1,79 @@
+[
+  {
+    "parentObject": "company",
+    "childObject": "opportunity",
+    "relationField": "companyId",
+    "childFilters": [
+      {
+        "field": "amount.amountMicros",
+        "operator": "gt",
+        "value": 0
+      }
+    ],
+    "aggregations": [
+      {
+        "type": "SUM",
+        "childField": "amount.amountMicros",
+        "parentField": "totalPipelineAmount",
+        "currencyField": "amount.currencyCode"
+      },
+      {
+        "type": "COUNT",
+        "parentField": "totalOpportunityCount"
+      },
+      {
+        "type": "SUM",
+        "childField": "amount.amountMicros",
+        "parentField": "wonPipelineAmount",
+        "currencyField": "amount.currencyCode",
+        "filters": [
+          {
+            "field": "stage",
+            "operator": "equals",
+            "value": "CUSTOMER"
+          }
+        ]
+      },
+      {
+        "type": "COUNT",
+        "parentField": "wonOpportunityCount",
+        "filters": [
+          {
+            "field": "stage",
+            "operator": "equals",
+            "value": "CUSTOMER"
+          }
+        ]
+      },
+      {
+        "type": "SUM",
+        "childField": "amount.amountMicros",
+        "parentField": "openPipelineAmount",
+        "currencyField": "amount.currencyCode",
+        "filters": [
+          {
+            "field": "stage",
+            "operator": "notEquals",
+            "value": "CUSTOMER"
+          }
+        ]
+      },
+      {
+        "type": "COUNT",
+        "parentField": "openOpportunityCount",
+        "filters": [
+          {
+            "field": "stage",
+            "operator": "notEquals",
+            "value": "CUSTOMER"
+          }
+        ]
+      },
+      {
+        "type": "MAX",
+        "childField": "closeDate",
+        "parentField": "lastOpportunityCloseDate"
+      }
+    ]
+  }
+]
@@ -0,0 +1,39 @@
+{
+  "version": "0.0.1",
+  "license": "MIT",
+  "engines": {
+    "node": "^24.5.0",
+    "npm": "please-use-yarn",
+    "yarn": ">=4.0.2"
+  },
+  "packageManager": "yarn@4.9.2",
+  "$schema": "https://raw.githubusercontent.com/twentyhq/twenty/main/packages/twenty-cli/schemas/appManifest.schema.json",
+  "universalIdentifier": "2b308f8c-6ff8-4838-9880-aa0271dfd8d8",
+  "name": "Rollup engine",
+  "description": "Rollup engine",
+  "env": {
+    "TWENTY_API_KEY": {
+      "isSecret": true,
+      "value": "",
+      "description": "Workspace API key used by the rollup engine to call the Twenty REST API."
+    },
+    "TWENTY_API_BASE_URL": {
+      "isSecret": false,
+      "value": "",
+      "description": "Optional override for the REST base URL (defaults to https://app.twenty.com/rest)."
+    },
+    "ROLLUP_ENGINE_CONFIG": {
+      "isSecret": false,
+      "value": "",
+      "description": "Optional JSON override for rollup definitions. Leave blank to use the baked-in defaults."
+    }
+  },
+  "scripts": {
+    "smoke": "node --import tsx scripts/run-rollup-smoke.ts",
+    "setup:metadata": "node scripts/setup-company-rollup-fields.mjs"
+  },
+  "devDependencies": {
+    "dotenv": "^16.4.5",
+    "tsx": "^4.20.6"
+  }
+}
@@ -0,0 +1,154 @@
+import assert from 'node:assert/strict';
+
+import { main as runRollups } from '../serverlessFunctions/calculaterollups/src/index.ts';
+
+type Json = Record<string, unknown>;
+
+interface RequestLogEntry {
+  url: string;
+  method: string;
+  body?: unknown;
+}
+
+const now = new Date();
+const currentYear = now.getUTCFullYear();
+const previousYear = currentYear - 1;
+
+const mockOpportunities = [
+  {
+    id: 'opp-1',
+    companyId: 'company-1',
+    amount: { amountMicros: 150_000_000, currencyCode: 'USD' },
+    stage: 'CUSTOMER',
+    closeDate: `${currentYear}-01-10T12:00:00.000Z`,
+  },
+  {
+    id: 'opp-2',
+    companyId: 'company-1',
+    amount: { amountMicros: 90_000_000, currencyCode: 'USD' },
+    stage: 'PROPOSAL',
+    closeDate: `${currentYear}-03-05T18:00:00.000Z`,
+  },
+  {
+    id: 'opp-3',
+    companyId: 'company-1',
+    amount: { amountMicros: 60_000_000, currencyCode: 'USD' },
+    stage: 'SCREENING',
+    closeDate: `${previousYear}-12-15T09:00:00.000Z`,
+  },
+];
+
+const requestLog: RequestLogEntry[] = [];
+const updatePayloads: Array<{ id: string; payload: Json }> = [];
+
+const jsonResponse = (data: Json | Json[]) =>
+  new Response(JSON.stringify(data), {
+    status: 200,
+    headers: { 'Content-Type': 'application/json' },
+  });
+
+global.fetch = async (
+  rawUrl: string | URL,
+  init?: { method?: string; body?: unknown },
+): Promise<any> => {
+  const url = typeof rawUrl === 'string' ? new URL(rawUrl) : rawUrl;
+  const method = (init?.method ?? 'GET').toUpperCase();
+  requestLog.push({
+    url: url.toString(),
+    method,
+    body: init?.body ? safeParse(init.body) : undefined,
+  });
+
+  if (url.pathname.endsWith('/opportunities') && method === 'GET') {
+    const companyId = url.searchParams.get('filter[companyId]');
+    const items = companyId
+      ? mockOpportunities.filter((opportunity) => opportunity.companyId === companyId)
+      : mockOpportunities;
+    return jsonResponse({
+      data: {
+        opportunities: items,
+      },
+      pageInfo: {
+        hasNextPage: false,
+      },
+    });
+  }
+
+  if (url.pathname.includes('/companies/') && method === 'PATCH') {
+    const id = url.pathname.split('/').pop() ?? 'unknown';
+    const payload = safeParse(init?.body) ?? {};
+    updatePayloads.push({ id, payload });
+    return jsonResponse({ data: { companies: [{ id, ...payload }] } });
+  }
+
+  throw new Error(`Unhandled request in mock fetch: ${method} ${url.toString()}`);
+};
+
+function safeParse(body: unknown) {
+  if (!body) {
+    return undefined;
+  }
+  if (typeof body === 'string') {
+    try {
+      return JSON.parse(body) as Json;
+    } catch (error) {
+      console.warn('Failed to parse request body', error);
+      return undefined;
+    }
+  }
+  return undefined;
+}
+
+async function main() {
+  process.env.TWENTY_API_KEY = 'mock-api-key';
+  process.env.TWENTY_API_BASE_URL = 'https://mock.twenty/api';
+
+  const params = {
+    trigger: { type: 'databaseEvent' },
+    record: { companyId: 'company-1' },
+    opportunity: { companyId: 'company-1' },
+  };
+
+  const result = await runRollups(params);
+
+  assert.equal(result.status, 'ok', 'rollup execution should succeed');
+
+  const expectedPayload = {
+    totalPipelineAmount: { amountMicros: 300_000_000, currencyCode: 'USD' },
+    totalOpportunityCount: 3,
+    wonPipelineAmount: { amountMicros: 150_000_000, currencyCode: 'USD' },
+    wonOpportunityCount: 1,
+    openPipelineAmount: { amountMicros: 150_000_000, currencyCode: 'USD' },
+    openOpportunityCount: 2,
+    lastOpportunityCloseDate: `${currentYear.toString().padStart(4, '0')}-03-05`,
+  };
+
+  const companyUpdate = updatePayloads.find((payload) => payload.id === 'company-1');
+  assert(companyUpdate, 'expected a PATCH payload for company-1');
+  assert.deepStrictEqual(companyUpdate.payload, expectedPayload);
+
+  const opportunitiesRequest = requestLog.find(
+    (entry) => entry.method === 'GET' && entry.url.includes('/opportunities'),
+  );
+  assert(opportunitiesRequest, 'expected at least one GET request for /opportunities');
+  const filterParams = new URL(opportunitiesRequest.url).searchParams.getAll('filter');
+  assert.deepStrictEqual(
+    filterParams,
+    ['companyId[eq]:"company-1"'],
+    'expected opportunity filter to use the new syntax',
+  );
+
+  console.log('--- Rollup execution summary ---');
+  console.dir(result, { depth: null });
+
+  console.log('\n--- PATCH payloads sent to companies ---');
+  console.dir(updatePayloads, { depth: null });
+
+  console.log('\n--- Requests made ---');
+  console.dir(requestLog, { depth: null });
+}
+
+main().catch((error) => {
+  console.error('Smoke test failed', error);
+  process.exitCode = 1;
+});
@@ -0,0 +1,138 @@
+import { config as loadEnv } from 'dotenv';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const __filename = fileURLToPath(import.meta.url);
+const __dirname = path.dirname(__filename);
+
+const envCandidates = [
+  process.env.TWENTY_ENV_PATH ? path.resolve(process.env.TWENTY_ENV_PATH) : null,
+  path.resolve(__dirname, '..', '.env'),
+  path.resolve(__dirname, '..', '..', '..', '..', '..', '.env'),
+].filter(Boolean);
+
+envCandidates.forEach((candidate) => {
+  if (!candidate) {
+    return;
+  }
+  const result = loadEnv({ path: candidate });
+  if (!result.error && process.env.TWENTY_API_KEY) {
+    return;
+  }
+});
+
+const METADATA_BASE_URL = (
+  process.env.TWENTY_METADATA_BASE_URL ??
+  (process.env.TWENTY_API_BASE_URL
+    ? `${process.env.TWENTY_API_BASE_URL.replace(/\/+$/, '')}/metadata`
+    : null) ??
+  'http://localhost:3000/rest/metadata'
+).replace(/\/+$/, '');
+
+const API_KEY = process.env.TWENTY_API_KEY;
+
+if (!API_KEY) {
+  console.error('TWENTY_API_KEY is required in the environment to call the Metadata API.');
+  process.exit(1);
+}
+
+const COMPANY_ROLLUP_FIELDS = [
+  { name: 'totalPipelineAmount', label: 'Total Pipeline Amount', type: 'CURRENCY' },
+  { name: 'totalOpportunityCount', label: 'Total Opportunity Count', type: 'NUMBER' },
+  { name: 'wonPipelineAmount', label: 'Won Pipeline Amount', type: 'CURRENCY' },
+  { name: 'wonOpportunityCount', label: 'Won Opportunity Count', type: 'NUMBER' },
+  { name: 'openPipelineAmount', label: 'Open Pipeline Amount', type: 'CURRENCY' },
+  { name: 'openOpportunityCount', label: 'Open Opportunity Count', type: 'NUMBER' },
+  { name: 'lastOpportunityCloseDate', label: 'Last Opportunity Close Date', type: 'DATE' },
+];
+
+const metadataRequest = async (method, endpoint, body) => {
+  const url = `${METADATA_BASE_URL}${endpoint}`;
+  const headers = {
+    Authorization: `Bearer ${API_KEY}`,
+  };
+  const init = { method, headers };
+  if (body !== undefined) {
+    headers['Content-Type'] = 'application/json';
+    init.body = JSON.stringify(body);
+  }
+
+  const response = await fetch(url, init);
+  const text = await response.text();
+  let parsed = {};
+  if (text) {
+    try {
+      parsed = JSON.parse(text);
+    } catch (error) {
+      throw new Error(`Failed to parse JSON response from ${url}: ${text}`);
+    }
+  }
+
+  if (!response.ok) {
+    const messages = parsed?.messages || parsed?.message;
+    const error = new Error(
+      `Metadata API ${method} ${endpoint} failed: ${messages || response.statusText}`,
+    );
+    error.response = parsed;
+    throw error;
+  }
+
+  return parsed;
+};
+
+const findObjectId = async (nameSingular) => {
+  const response = await metadataRequest('GET', '/objects');
+  const objects =
+    (response?.data && Array.isArray(response.data.objects) ? response.data.objects : []) || [];
+  const match = objects.find((entry) => entry.nameSingular === nameSingular);
+  if (!match) {
+    throw new Error(`Unable to find object with nameSingular="${nameSingular}" via Metadata API`);
+  }
+  return match.id;
+};
+
+const createField = async (objectMetadataId, field) => {
+  try {
+    const payload = {
+      objectMetadataId,
+      name: field.name,
+      label: field.label,
+      type: field.type,
+    };
+    await metadataRequest('POST', '/fields', payload);
+    console.log(`✔ Created ${field.name}`);
+  } catch (error) {
+    const messages = error.response?.messages;
+    const messageList = Array.isArray(messages)
+      ? messages
+      : typeof messages === 'string'
+        ? [messages]
+        : [];
+    if (messageList.some((msg) => msg.includes('already exists'))) {
+      console.log(`→ ${field.name} already exists; skipping`);
+      return;
+    }
+    throw error;
+  }
+};
+
+const main = async () => {
+  console.log(`Using metadata base URL: ${METADATA_BASE_URL}`);
+  const companyObjectId = await findObjectId('company');
+
+  for (const field of COMPANY_ROLLUP_FIELDS) {
+    // Metadata API prefers camelCase names; script assumes labels may contain spaces.
+    await createField(companyObjectId, field);
+  }
+
+  console.log('Company rollup fields ensured.');
+};
+
+main().catch((error) => {
+  console.error('Failed to provision rollup fields via Metadata API.');
+  console.error(error instanceof Error ? error.message : error);
+  if (error instanceof Error && error.stack) {
+    console.error(error.stack);
+  }
+  process.exit(1);
+});
@@ -0,0 +1,24 @@
+{
+  "$schema": "https://raw.githubusercontent.com/twentyhq/twenty/main/packages/twenty-cli/schemas/serverlessFunction.schema.json",
+  "universalIdentifier": "c3ec36c8-5b1d-421f-9172-a9e035ab9c18",
+  "name": "calculaterollups",
+  "triggers": [
+    {
+      "universalIdentifier": "eec8aaf2-b0cc-47fd-b522-8d4aa5fe4bd3",
+      "type": "databaseEvent",
+      "eventName": "opportunity.*"
+    },
+    {
+      "universalIdentifier": "a3fea230-1121-44a6-b395-5811c3031f8e",
+      "type": "cron",
+      "schedule": " 0 2 * * *"
+    },
+    {
+      "universalIdentifier": "d33b0fe4-4b2b-45c0-aa2f-e617fdbba484",
+      "type": "route",
+      "path": "/recalculate-all",
+      "httpMethod": "POST",
+      "isAuthRequired": true
+    }
+  ]
+}
@@ -0,0 +1,128 @@
+import { applyFilters, getNestedValue, toComparableNumber } from './filtering';
+import type { ChildRecord, RollupDefinition } from './types';
+
+const roundForSum = (value: number) =>
+  Number.isInteger(value) ? value : Math.round(value * 100) / 100;
+
+export const computeAggregations = (
+  definition: RollupDefinition,
+  records: ChildRecord[],
+  now: Date,
+) => {
+  const baseFiltered = applyFilters(records, definition.childFilters, now);
+  const result: Record<string, number | string | null | Record<string, unknown>> = {};
+
+  definition.aggregations.forEach((aggregation) => {
+    const scopedRecords = applyFilters(baseFiltered, aggregation.filters, now);
+
+    switch (aggregation.type) {
+      case 'COUNT':
+        result[aggregation.parentField] = scopedRecords.length;
+        break;
+      case 'SUM': {
+        if (!aggregation.childField) {
+          throw new Error('SUM aggregation requires childField');
+        }
+        const total = scopedRecords.reduce(
+          (accumulator, record) => {
+            const rawValue = getNestedValue(record, aggregation.childField!);
+            const currencyRaw =
+              aggregation.currencyField !== undefined
+                ? getNestedValue(record, aggregation.currencyField)
+                : undefined;
+            const numeric =
+              typeof rawValue === 'number'
+                ? rawValue
+                : typeof rawValue === 'string'
+                  ? Number(rawValue)
+                  : NaN;
+            if (Number.isNaN(numeric)) {
+              return accumulator;
+            }
+            return {
+              amount: accumulator.amount + numeric,
+              currency:
+                typeof currencyRaw === 'string' && currencyRaw.trim().length > 0
+                  ? currencyRaw
+                  : accumulator.currency,
+            };
+          },
+          { amount: 0, currency: undefined as string | undefined },
+        );
+        result[aggregation.parentField] = {
+          amountMicros: Math.round(roundForSum(total.amount)),
+          currencyCode: total.currency ?? '',
+        };
+        break;
+      }
+      case 'AVG': {
+        if (!aggregation.childField) {
+          throw new Error('AVG aggregation requires childField');
+        }
+        const { total, count } = scopedRecords.reduce(
+          (accumulator, record) => {
+            const rawValue = getNestedValue(record, aggregation.childField!);
+            const numeric =
+              typeof rawValue === 'number'
+                ? rawValue
+                : typeof rawValue === 'string'
+                  ? Number(rawValue)
+                  : NaN;
+            if (Number.isNaN(numeric)) {
+              return accumulator;
+            }
+            return {
+              total: accumulator.total + numeric,
+              count: accumulator.count + 1,
+            };
+          },
+          { total: 0, count: 0 },
+        );
+        result[aggregation.parentField] = count === 0 ? null : roundForSum(total / count);
+        break;
+      }
+      case 'MAX':
+      case 'MIN': {
+        if (!aggregation.childField) {
+          throw new Error(`${aggregation.type} aggregation requires childField`);
+        }
+        const direction = aggregation.type === 'MAX' ? 1 : -1;
+        let chosen: { raw: unknown; comparable: number | null } | null = null;
+        scopedRecords.forEach((record) => {
+          const rawValue = getNestedValue(record, aggregation.childField!);
+          const comparable = toComparableNumber(rawValue);
+          if (comparable === null) {
+            return;
+          }
+          if (
+            chosen === null ||
+            (chosen.comparable !== null && direction * (comparable - chosen.comparable) > 0)
+          ) {
+            chosen = { raw: rawValue, comparable };
+          }
+        });
+
+        if (chosen === null) {
+          result[aggregation.parentField] = null;
+        } else if (typeof chosen.raw === 'number') {
+          result[aggregation.parentField] = chosen.raw;
+        } else if (chosen.raw instanceof Date) {
+          result[aggregation.parentField] = chosen.raw.toISOString().slice(0, 10);
+        } else if (chosen.raw === null || chosen.raw === undefined) {
+          result[aggregation.parentField] = null;
+        } else {
+          const rawString = String(chosen.raw);
+          const parsed = Date.parse(rawString);
+          result[aggregation.parentField] = Number.isNaN(parsed)
+            ? rawString
+            : new Date(parsed).toISOString().slice(0, 10);
+        }
+        break;
+      }
+      default:
+        break;
+    }
+  });
+
+  return result;
+};
@@ -0,0 +1,347 @@
+import { getNestedValue } from './filtering';
+import type { ChildRecord, RollupDefinition } from './types';
+
+const RESOURCE_PLURALS: Record<string, string> = {
+  person: 'people',
+  gift: 'gifts',
+  company: 'companies',
+  opportunity: 'opportunities',
+};
+
+const RETRIABLE_STATUS = new Set([429, 500, 502, 503, 504]);
+const DEFAULT_PAGE_SIZE = 200;
+const MAX_RETRIES = 3;
+const RETRY_BASE_DELAY_MS = 300;
+
+type QueryParamPrimitive = string | number | boolean;
+type QueryParamValue = QueryParamPrimitive | QueryParamPrimitive[];
+
+const serializeFilterExpressions = (
+  filters: Record<string, QueryParamValue>,
+): string[] => {
+  const expressions: string[] = [];
+
+  Object.entries(filters).forEach(([field, rawValue]) => {
+    if (rawValue === undefined || rawValue === null) {
+      return;
+    }
+
+    if (Array.isArray(rawValue)) {
+      if (rawValue.length === 0) {
+        return;
+      }
+      const serializedValues = rawValue.map((entry) => JSON.stringify(entry)).join(',');
+      expressions.push(`${field}[in]:${serializedValues}`);
+      return;
+    }
+
+    expressions.push(`${field}[eq]:${JSON.stringify(rawValue)}`);
+  });
+
+  return expressions;
+};
+
+interface RequestOptions {
+  method?: 'GET' | 'POST' | 'PATCH' | 'PUT' | 'DELETE';
+  query?: Record<string, QueryParamValue>;
+  body?: unknown;
+  headers?: Record<string, string>;
+}
+
+interface ListPageOptions {
+  filter?: Record<string, QueryParamValue>;
+  orderBy?: Record<string, 'asc' | 'desc'>;
+  cursor?: string;
+  limit?: number;
+}
+
+interface ListPage<T> {
+  items: T[];
+  hasNextPage: boolean;
+  nextCursor?: string;
+}
+
+const sleep = (ms: number) =>
+  new Promise((resolve) => {
+    setTimeout(resolve, ms);
+  });
+
+const getResourceName = (objectName: string) => RESOURCE_PLURALS[objectName] ?? `${objectName}s`;
+
+export class TwentyClient {
+  private readonly baseUrl: string;
+
+  constructor(
+    private readonly apiKey: string,
+    baseUrl: string,
+  ) {
+    this.baseUrl = baseUrl.replace(/\/+$/, '');
+  }
+
+  private async fetchWithRetry(url: string, init: RequestInit, attempt = 1): Promise<Response> {
+    const response = await fetch(url, init);
+    if (response.ok) {
+      return response;
+    }
+
+    if (RETRIABLE_STATUS.has(response.status) && attempt < MAX_RETRIES) {
+      const delayMs = RETRY_BASE_DELAY_MS * attempt;
+      await sleep(delayMs);
+      return this.fetchWithRetry(url, init, attempt + 1);
+    }
+
+    const errorBody = await response.text();
+    throw new Error(`Request failed (${response.status}): ${errorBody || 'no body returned'}`);
+  }
+
+  private appendQueryParams(url: URL, params: Record<string, QueryParamValue>) {
+    Object.entries(params).forEach(([key, value]) => {
+      if (value === undefined || value === null) {
+        return;
+      }
+      if (Array.isArray(value)) {
+        value.forEach((entry) => {
+          url.searchParams.append(key, String(entry));
+        });
+        return;
+      }
+      url.searchParams.set(key, String(value));
+    });
+  }
+
+  private async request(resourcePath: string, options: RequestOptions = {}) {
+    const { method = 'GET', query, body, headers = {} } = options;
+
+    const url = resourcePath.startsWith('http')
+      ? new URL(resourcePath)
+      : new URL(`${this.baseUrl}/${resourcePath.replace(/^\/+/, '')}`);
+
+    if (query) {
+      this.appendQueryParams(url, query);
+    }
+
+    let serializedBody: string | undefined;
+
+    if (body !== undefined) {
+      serializedBody = typeof body === 'string' ? body : JSON.stringify(body);
+    }
+
+    const mergedHeaders: Record<string, string> = {
+      Authorization: `Bearer ${this.apiKey}`,
+      ...headers,
+    };
+
+    if (serializedBody && !mergedHeaders['Content-Type']) {
+      mergedHeaders['Content-Type'] = 'application/json';
+    }
+
+    const response = await this.fetchWithRetry(url.toString(), {
+      method,
+      headers: mergedHeaders,
+      body: serializedBody,
+    });
+
+    if (response.status === 204) {
+      return {};
+    }
+
+    const text = await response.text();
+    if (!text) {
+      return {};
+    }
+
+    try {
+      return JSON.parse(text) as unknown;
+    } catch (error) {
+      throw new Error(
+        `Failed to parse JSON response from ${url.toString()}: ${(error as Error).message}`,
+      );
+    }
+  }
+
+  private extractRecords(body: unknown, resource: string): ChildRecord[] {
+    if (typeof body !== 'object' || body === null) {
+      return [];
+    }
+    const data = typeof (body as Record<string, unknown>).data === 'object'
+      ? ((body as Record<string, unknown>).data as Record<string, unknown>)
+      : undefined;
+
+    const direct = data && Array.isArray(data[resource]) ? data[resource] : undefined;
+    if (Array.isArray(direct)) {
+      return direct as ChildRecord[];
+    }
+
+    const singular = resource.endsWith('s') ? resource.slice(0, -1) : resource;
+    const singularMatch = data && Array.isArray(data[singular]) ? data[singular] : undefined;
+    if (Array.isArray(singularMatch)) {
+      return singularMatch as ChildRecord[];
+    }
+
+    const capitalized = resource.charAt(0).toUpperCase() + resource.slice(1);
+    const findManyKey = `findMany${capitalized}`;
+    const findMany = data && Array.isArray(data[findManyKey]) ? data[findManyKey] : undefined;
+    if (Array.isArray(findMany)) {
+      return findMany as ChildRecord[];
+    }
+
+    return [];
+  }
+
+  private extractPageInfo(body: unknown) {
+    if (typeof body !== 'object' || body === null) {
+      return undefined;
+    }
+    if (typeof (body as Record<string, unknown>).pageInfo === 'object') {
+      return (body as Record<string, unknown>).pageInfo as Record<string, unknown>;
+    }
+    if (
+      typeof (body as Record<string, unknown>).data === 'object' &&
+      typeof ((body as Record<string, unknown>).data as Record<string, unknown>).pageInfo === 'object'
+    ) {
+      return ((body as Record<string, unknown>).data as Record<string, unknown>)
+        .pageInfo as Record<string, unknown>;
+    }
+    return undefined;
+  }
+
+  private async listRecordsPage(
+    resource: string,
+    options: ListPageOptions = {},
+  ): Promise<ListPage<ChildRecord>> {
+    const query: Record<string, QueryParamValue> = {};
+    const limit = options.limit ?? DEFAULT_PAGE_SIZE;
+    query.limit = String(limit);
+
+    if (options.cursor) {
+      query.starting_after = options.cursor;
+    }
+
+    if (options.filter) {
+      const filterExpressions = serializeFilterExpressions(options.filter);
+      if (filterExpressions.length > 0) {
+        query.filter = filterExpressions;
+      }
+    }
+
+    if (options.orderBy) {
+      Object.entries(options.orderBy).forEach(([field, direction]) => {
+        query[`order_by[${field}]`] = direction;
+      });
+    }
+
+    const response = await this.request(resource, {
+      method: 'GET',
+      query,
+    });
+
+    const items = this.extractRecords(response, resource);
+    const pageInfo = this.extractPageInfo(response);
+    const hasNextPage =
+      !!pageInfo &&
+      (Boolean(pageInfo.hasNextPage) ||
+        Boolean(pageInfo.endCursor) ||
+        Boolean(pageInfo.nextCursor));
+    const nextCursor =
+      (pageInfo && typeof pageInfo.endCursor === 'string' && pageInfo.endCursor) ||
+      (pageInfo && typeof pageInfo.nextCursor === 'string' && pageInfo.nextCursor) ||
+      undefined;
+
+    return {
+      items,
+      hasNextPage,
+      nextCursor,
+    };
+  }
+
+  async listAllRecords(objectName: string, options: ListPageOptions = {}): Promise<ChildRecord[]> {
+    const resource = getResourceName(objectName);
+    const aggregated: ChildRecord[] = [];
+    let cursor: string | undefined;
+    let hasNext = true;
+
+    while (hasNext) {
+      const page = await this.listRecordsPage(resource, {
+        ...options,
+        cursor,
+      });
+      aggregated.push(...page.items);
+      hasNext = page.hasNextPage && Boolean(page.nextCursor);
+      cursor = hasNext ? page.nextCursor : undefined;
+    }
+
+    return aggregated;
+  }
+
+  async updateObject(objectName: string, id: string, payload: Record<string, unknown>) {
+    const resource = getResourceName(objectName);
+    await this.request(`${resource}/${id}`, {
+      method: 'PATCH',
+      body: payload,
+    });
+  }
+}
+
+export const ensureMapHasTargets = (
+  container: Map<string, ChildRecord[]>,
+  targetIds: Set<string> | undefined,
+) => {
+  if (!targetIds) {
+    return;
+  }
+  targetIds.forEach((id) => {
+    if (!container.has(id)) {
+      container.set(id, []);
+    }
+  });
+};
+
+export const buildChildRecordIndex = async (
+  definition: RollupDefinition,
+  client: TwentyClient,
+  parentIds: Set<string> | undefined,
+): Promise<Map<string, ChildRecord[]>> => {
+  if (parentIds && parentIds.size > 0) {
+    const parentIdList = Array.from(parentIds);
+    const result = new Map<string, ChildRecord[]>();
+    const concurrency = 5;
+
+    for (let index = 0; index < parentIdList.length; index += concurrency) {
+      const slice = parentIdList.slice(index, index + concurrency);
+      const batch = await Promise.all(
+        slice.map(async (parentId) => {
+          const records = await client.listAllRecords(definition.childObject, {
+            filter: {
+              [definition.relationField]: parentId,
+            },
+          });
+          const filtered = records.filter((record) => {
+            const relationValue = getNestedValue(record, definition.relationField);
+            return typeof relationValue === 'string' && relationValue === parentId;
+          });
+          return { parentId, records: filtered };
+        }),
+      );
+      batch.forEach(({ parentId, records }) => {
+        result.set(parentId, records);
+      });
+    }
+
+    ensureMapHasTargets(result, parentIds);
+    return result;
+  }
+
+  const allRecords = await client.listAllRecords(definition.childObject);
+  const grouped = new Map<string, ChildRecord[]>();
+  allRecords.forEach((record) => {
+    const relationValue = getNestedValue(record, definition.relationField);
+    if (typeof relationValue !== 'string' || relationValue.trim().length === 0) {
+      return;
+    }
+    if (!grouped.has(relationValue)) {
+      grouped.set(relationValue, []);
+    }
+    grouped.get(relationValue)!.push(record);
+  });
+  return grouped;
+};
@@ -0,0 +1,174 @@
+import { defaultRollupConfig } from './rollupConfig';
+import type {
+  AggregationConfig,
+  FilterConfig,
+  RollupConfig,
+  RollupDefinition,
+} from './types';
+
+const operatorSet = new Set<FilterConfig['operator']>([
+  'equals',
+  'notEquals',
+  'in',
+  'notIn',
+  'gt',
+  'gte',
+  'lt',
+  'lte',
+]);
+
+const aggregationTypes = new Set<AggregationConfig['type']>([
+  'SUM',
+  'COUNT',
+  'MAX',
+  'MIN',
+  'AVG',
+]);
+
+const isObject = (value: unknown): value is Record<string, unknown> =>
+  typeof value === 'object' && value !== null;
+
+export const validateRollupConfig = (config: unknown): asserts config is RollupConfig => {
+  if (!Array.isArray(config)) {
+    throw new Error('Rollup configuration must contain an array of rollup definitions');
+  }
+
+  config.forEach((definition, definitionIndex) => {
+    if (!isObject(definition)) {
+      throw new Error(`Definition at index ${definitionIndex} must be an object`);
+    }
+
+    const { parentObject, childObject, relationField, childFilters, aggregations } =
+      definition as Partial<RollupDefinition>;
+
+    if (typeof parentObject !== 'string' || parentObject.trim().length === 0) {
+      throw new Error(`Definition ${definitionIndex} missing parentObject`);
+    }
+    if (typeof childObject !== 'string' || childObject.trim().length === 0) {
+      throw new Error(`Definition ${definitionIndex} missing childObject`);
+    }
+    if (typeof relationField !== 'string' || relationField.trim().length === 0) {
+      throw new Error(`Definition ${definitionIndex} missing relationField`);
+    }
+    if (!Array.isArray(aggregations) || aggregations.length === 0) {
+      throw new Error(`Definition ${definitionIndex} must declare at least one aggregation`);
+    }
+
+    const filtersToValidate = [
+      ...(Array.isArray(childFilters) ? childFilters : []),
+      ...aggregations.flatMap((aggregation, aggregationIndex) => {
+        if (!isObject(aggregation)) {
+          throw new Error(
+            `Aggregation ${aggregationIndex} in definition ${definitionIndex} must be an object`,
+          );
+        }
+
+        const { type, parentField, childField, filters } = aggregation as AggregationConfig;
+
+        if (!aggregationTypes.has(type)) {
+          throw new Error(
+            `Aggregation ${aggregationIndex} in definition ${definitionIndex} has unsupported type ${type}`,
+          );
+        }
+
+        if (typeof parentField !== 'string' || parentField.trim().length === 0) {
+          throw new Error(
+            `Aggregation ${aggregationIndex} in definition ${definitionIndex} missing parentField`,
+          );
+        }
+
+        if (type !== 'COUNT' && (typeof childField !== 'string' || childField.length === 0)) {
+          throw new Error(
+            `Aggregation ${aggregationIndex} in definition ${definitionIndex} with type ${type} requires childField`,
+          );
+        }
+
+        if (filters && !Array.isArray(filters)) {
+          throw new Error(
+            `Aggregation ${aggregationIndex} in definition ${definitionIndex} has invalid filters shape`,
+          );
+        }
+
+        return filters ?? [];
+      }),
+    ];
+
+    filtersToValidate.forEach((filter, filterIndex) => {
+      if (!isObject(filter)) {
+        throw new Error(
+          `Filter ${filterIndex} in definition ${definitionIndex} must be an object`,
+        );
+      }
+
+      const { field, operator } = filter as FilterConfig;
+
+      if (typeof field !== 'string' || field.trim().length === 0) {
+        throw new Error(
+          `Filter ${filterIndex} in definition ${definitionIndex} missing field`,
+        );
+      }
+
+      if (!operatorSet.has(operator)) {
+        throw new Error(
+          `Filter ${filterIndex} in definition ${definitionIndex} has unsupported operator ${operator}`,
+        );
+      }
+    });
+  });
+};
+
+const collectValuesByKey = (
+  value: unknown,
+  key: string,
+  result: Set<string>,
+): void => {
+  if (Array.isArray(value)) {
+    value.forEach((entry) => collectValuesByKey(entry, key, result));
+    return;
+  }
+
+  if (!isObject(value)) {
+    return;
+  }
+
+  Object.entries(value).forEach(([currentKey, entryValue]) => {
+    if (currentKey === key && typeof entryValue === 'string' && entryValue.trim().length > 0) {
+      result.add(entryValue);
+      return;
+    }
+
+    collectValuesByKey(entryValue, key, result);
+  });
+};
+
+export const extractRelationValues = (
+  params: unknown,
+  relationField: string,
+): Set<string> => {
+  const values = new Set<string>();
+  collectValuesByKey(params, relationField, values);
+  return values;
+};
+
+export const resolveRollupConfig = (): RollupConfig => {
+  const override =
+    process.env.ROLLUP_ENGINE_CONFIG ??
+    process.env.ROLLUPS_CONFIG ??
+    process.env.CALCULATE_ROLLUPS_CONFIG;
+
+  if (override) {
+    try {
+      const parsed = JSON.parse(override) as unknown;
+      validateRollupConfig(parsed);
+      return parsed;
+    } catch (error) {
+      const reason =
+        error instanceof Error ? error.message : typeof error === 'string' ? error : 'Unknown error';
+      throw new Error(`Unable to parse rollup configuration override (reason: ${reason})`);
+    }
+  }
+
+  const config = defaultRollupConfig;
+  validateRollupConfig(config);
+  return config;
+};
@@ -0,0 +1,149 @@
+import type { ChildRecord, DynamicValue, FilterConfig } from './types';
+
+const isObject = (value: unknown): value is Record<string, unknown> =>
+  typeof value === 'object' && value !== null;
+
+const resolveDynamicValue = (dynamicValue: DynamicValue, now: Date) => {
+  switch (dynamicValue) {
+    case 'startOfYear': {
+      const utcStart = new Date(Date.UTC(now.getUTCFullYear(), 0, 1, 0, 0, 0, 0));
+      return utcStart.toISOString();
+    }
+    default:
+      throw new Error(`Unsupported dynamicValue ${dynamicValue}`);
+  }
+};
+
+const normalizeForEquality = (value: unknown) => {
+  if (value === null || value === undefined) {
+    return null;
+  }
+  if (typeof value === 'number' || typeof value === 'boolean') {
+    return value;
+  }
+  if (typeof value === 'string') {
+    const numeric = Number(value);
+    if (!Number.isNaN(numeric)) {
+      return numeric;
+    }
+    return value;
+  }
+  if (value instanceof Date) {
+    return value.toISOString();
+  }
+  return null;
+};
+
+export const toComparableNumber = (value: unknown): number | null => {
+  if (typeof value === 'number') {
+    return value;
+  }
+  if (typeof value === 'string') {
+    const numeric = Number(value);
+    if (!Number.isNaN(numeric)) {
+      return numeric;
+    }
+    const timestamp = Date.parse(value);
+    if (!Number.isNaN(timestamp)) {
+      return timestamp;
+    }
+    return null;
+  }
+  if (value instanceof Date) {
+    return value.getTime();
+  }
+  return null;
+};
+
+const compareValues = (left: unknown, right: unknown): number | null => {
+  const leftComparable = toComparableNumber(left);
+  const rightComparable = toComparableNumber(right);
+  if (leftComparable === null || rightComparable === null) {
+    return null;
+  }
+  return leftComparable - rightComparable;
+};
+
+const evaluateFilter = (rawValue: unknown, filter: FilterConfig, now: Date) => {
+  const comparisonSource =
+    filter.dynamicValue !== undefined
+      ? resolveDynamicValue(filter.dynamicValue, now)
+      : filter.value;
+
+  switch (filter.operator) {
+    case 'equals': {
+      const left = normalizeForEquality(rawValue);
+      const right = normalizeForEquality(comparisonSource);
+      return left !== null && right !== null ? left === right : rawValue === comparisonSource;
+    }
+    case 'notEquals': {
+      const left = normalizeForEquality(rawValue);
+      const right = normalizeForEquality(comparisonSource);
+      return left !== null && right !== null ? left !== right : rawValue !== comparisonSource;
+    }
+    case 'in': {
+      if (!Array.isArray(comparisonSource)) {
+        return false;
+      }
+      const left = normalizeForEquality(rawValue);
+      if (left === null) {
+        return false;
+      }
+      return comparisonSource.map(normalizeForEquality).some((candidate) => candidate === left);
+    }
+    case 'notIn': {
+      if (!Array.isArray(comparisonSource)) {
+        return true;
+      }
+      const left = normalizeForEquality(rawValue);
+      if (left === null) {
+        return true;
+      }
+      return !comparisonSource.map(normalizeForEquality).some((candidate) => candidate === left);
+    }
+    case 'gt': {
+      const comparison = compareValues(rawValue, comparisonSource);
+      return comparison !== null ? comparison > 0 : false;
+    }
+    case 'gte': {
+      const comparison = compareValues(rawValue, comparisonSource);
+      return comparison !== null ? comparison >= 0 : false;
+    }
+    case 'lt': {
+      const comparison = compareValues(rawValue, comparisonSource);
+      return comparison !== null ? comparison < 0 : false;
+    }
+    case 'lte': {
+      const comparison = compareValues(rawValue, comparisonSource);
+      return comparison !== null ? comparison <= 0 : false;
+    }
+    default:
+      return true;
+  }
+};
+
+export const getNestedValue = (record: ChildRecord, pathExpression: string): unknown => {
+  if (!pathExpression.includes('.')) {
+    return record[pathExpression];
+  }
+
+  return pathExpression.split('.').reduce<unknown>((accumulator, key) => {
+    if (!isObject(accumulator)) {
+      return undefined;
+    }
+    return accumulator[key];
+  }, record);
+};
+
+export const applyFilters = (
+  records: ChildRecord[],
+  filters: FilterConfig[] | undefined,
+  now: Date,
+) => {
+  if (!filters || filters.length === 0) {
+    return records;
+  }
+  return records.filter((record) =>
+    filters.every((filter) => evaluateFilter(getNestedValue(record, filter.field), filter, now)),
+  );
+};
@@ -0,0 +1,207 @@
+import { computeAggregations } from './aggregations';
+import { buildChildRecordIndex, TwentyClient } from './client';
+import { extractRelationValues, resolveRollupConfig } from './config';
+import { getNestedValue } from './filtering';
+import type { ExecutionSummaryItem, RollupDefinition } from './types';
+
+const isObject = (value: unknown): value is Record<string, unknown> =>
+  typeof value === 'object' && value !== null;
+
+const sanitizePayload = (payload: Record<string, unknown>) =>
+  Object.fromEntries(
+    Object.entries(payload).filter(([, value]) => {
+      if (value === undefined) {
+        return false;
+      }
+      if (typeof value === 'number' && !Number.isFinite(value)) {
+        return false;
+      }
+      return true;
+    }),
+  );
+
+const determineFullRebuild = (params: unknown) => {
+  if (!isObject(params)) {
+    return false;
+  }
+  if (params.recalculateAll === true || params.fullRebuild === true) {
+    return true;
+  }
+  if (isObject(params.trigger) && params.trigger.type === 'cron') {
+    return true;
+  }
+  if (params.trigger === 'cron') {
+    return true;
+  }
+  return false;
+};
+
+const getApiCredentials = () => {
+  const apiKey =
+    process.env.TWENTY_API_KEY ??
+    process.env.TWENTY_API_TOKEN ??
+    process.env.API_KEY;
+
+  if (!apiKey) {
+    return null;
+  }
+
+  const baseUrlRaw =
+    process.env.TWENTY_API_BASE_URL ??
+    process.env.TWENTY_REST_BASE_URL ??
+    process.env.TWENTY_API_URL ??
+    '';
+
+  const baseUrl =
+    typeof baseUrlRaw === 'string' && baseUrlRaw.trim().length > 0
+      ? baseUrlRaw
+      : 'https://app.twenty.com/rest';
+
+  return { apiKey, baseUrl };
+};
+
+const formatSummary = (summaries: ExecutionSummaryItem[], durationMs: number) => ({
+  status: 'ok',
+  tookMs: durationMs,
+  totals: {
+    processed: summaries.reduce((accumulator, item) => accumulator + item.processed, 0),
+    updated: summaries.reduce((accumulator, item) => accumulator + item.updated, 0),
+  },
+  details: summaries,
+});
+
+export async function main(params: unknown): Promise<object> {
+  const start = Date.now();
+  try {
+    const config = resolveRollupConfig();
+    if (config.length === 0) {
+      return { status: 'noop', reason: 'No rollup definitions configured' };
+    }
+
+    const fullRebuild = determineFullRebuild(params);
+    const relationCache = new Map<string, Set<string>>();
+
+    config.forEach((definition) => {
+      if (!relationCache.has(definition.relationField)) {
+        relationCache.set(
+          definition.relationField,
+          extractRelationValues(params, definition.relationField),
+        );
+      }
+    });
+
+    const credentials = getApiCredentials();
+    if (!credentials) {
+      console.warn('[rollup] skipping execution because TWENTY_API_KEY is not set');
+      return { status: 'noop', reason: 'TWENTY_API_KEY not configured' };
+    }
+
+    const { apiKey, baseUrl } = credentials;
+    const client = new TwentyClient(apiKey, baseUrl);
+    const now = new Date();
+    const summaries: ExecutionSummaryItem[] = [];
+
+    for (const definition of config) {
+      const targetIds = fullRebuild ? undefined : relationCache.get(definition.relationField);
+
+      if (!fullRebuild && (!targetIds || targetIds.size === 0)) {
+        summaries.push({
+          parentObject: definition.parentObject,
+          processed: 0,
+          updated: 0,
+          relationField: definition.relationField,
+          mode: 'targeted',
+          skipped: `No ${definition.relationField} values found in payload`,
+        });
+        continue;
+      }
+
+      const childIndex = await buildChildRecordIndex(definition, client, targetIds);
+
+      const updates: Array<{
+        id: string;
+        payload: Record<string, unknown>;
+        context: { relationId: string };
+      }> = [];
+      childIndex.forEach((records, parentId) => {
+        const recordIds = records
+          .map((record) => getNestedValue(record, 'id'))
+          .filter((value): value is string => typeof value === 'string');
+        console.info(
+          `[rollup] relation ${parentId} includes ${recordIds.length} ${definition.childObject}(s): ${recordIds.join(', ')}`,
+        );
+        const aggregates = computeAggregations(definition, records, now);
+        const payload = sanitizePayload(aggregates);
+        if (Object.keys(payload).length === 0) {
+          return;
+        }
+        console.info(
+          `[rollup] computed aggregates for ${definition.parentObject} ${parentId}: ${JSON.stringify(payload)}`,
+        );
+        updates.push({ id: parentId, payload, context: { relationId: parentId } });
+      });
+
+      let updatedCount = 0;
+      for (const update of updates) {
+        try {
+          await client.updateObject(definition.parentObject, update.id, update.payload);
+          updatedCount += 1;
+          console.info(
+            `[rollup] updated ${definition.parentObject} ${update.id} (relation ${update.context.relationId})`,
+          );
+        } catch (error) {
+          const reason =
+            error instanceof Error
+              ? error.message || error.toString()
+              : typeof error === 'string'
+                ? error
+                : 'Unknown error';
+          console.warn(
+            `[rollup] failed to update ${definition.parentObject} ${update.id} (relation ${update.context.relationId}): ${reason}`,
+          );
+        }
+      }
+
+      summaries.push({
+        parentObject: definition.parentObject,
+        processed: childIndex.size,
+        updated: updatedCount,
+        relationField: definition.relationField,
+        mode: fullRebuild ? 'full-rebuild' : 'targeted',
+      });
+    }
+
+    const totalProcessed = summaries.reduce(
+      (accumulator, item) => accumulator + item.processed,
+      0,
+    );
+
+    if (!fullRebuild && totalProcessed === 0) {
+      return {
+        status: 'noop',
+        reason: 'No matching relation ids found in payload',
+      };
+    }
+
+    const duration = Date.now() - start;
+    console.info(
+      `[rollup] completed mode=${fullRebuild ? 'full-rebuild' : 'targeted'} processed=${totalProcessed} durationMs=${duration}`,
+    );
+    return formatSummary(summaries, duration);
+  } catch (error) {
+    const serializedError =
+      error instanceof Error
+        ? `${error.name}: ${error.message}${error.stack ? `\n${error.stack}` : ''}`
+        : JSON.stringify(error);
+    console.error('[rollup] execution failed', serializedError);
+    return {
+      status: 'error',
+      message:
+        error instanceof Error
+          ? error.message || 'Unknown error'
+          : typeof error === 'string'
+            ? error
+            : 'Unknown error',
+    };
+  }
+}
@@ -0,0 +1,81 @@
+import type { RollupConfig } from './types';
+
+export const defaultRollupConfig: RollupConfig = [
+  {
+    parentObject: 'company',
+    childObject: 'opportunity',
+    relationField: 'companyId',
+    childFilters: [
+      {
+        field: 'amount.amountMicros',
+        operator: 'gt',
+        value: 0,
+      },
+    ],
+    aggregations: [
+      {
+        type: 'SUM',
+        childField: 'amount.amountMicros',
+        parentField: 'totalPipelineAmount',
+        currencyField: 'amount.currencyCode',
+      },
+      {
+        type: 'COUNT',
+        parentField: 'totalOpportunityCount',
+      },
+      {
+        type: 'SUM',
+        childField: 'amount.amountMicros',
+        parentField: 'wonPipelineAmount',
+        currencyField: 'amount.currencyCode',
+        filters: [
+          {
+            field: 'stage',
+            operator: 'equals',
+            value: 'CUSTOMER',
+          },
+        ],
+      },
+      {
+        type: 'COUNT',
+        parentField: 'wonOpportunityCount',
+        filters: [
+          {
+            field: 'stage',
+            operator: 'equals',
+            value: 'CUSTOMER',
+          },
+        ],
+      },
+      {
+        type: 'SUM',
+        childField: 'amount.amountMicros',
+        parentField: 'openPipelineAmount',
+        currencyField: 'amount.currencyCode',
+        filters: [
+          {
+            field: 'stage',
+            operator: 'notEquals',
+            value: 'CUSTOMER',
+          },
+        ],
+      },
+      {
+        type: 'COUNT',
+        parentField: 'openOpportunityCount',
+        filters: [
+          {
+            field: 'stage',
+            operator: 'notEquals',
+            value: 'CUSTOMER',
+          },
+        ],
+      },
+      {
+        type: 'MAX',
+        childField: 'closeDate',
+        parentField: 'lastOpportunityCloseDate',
+      },
+    ],
+  },
+];
@@ -0,0 +1,49 @@
+export type AggregationType = 'SUM' | 'COUNT' | 'MAX' | 'MIN' | 'AVG';
+
+export type Operator =
+  | 'equals'
+  | 'notEquals'
+  | 'in'
+  | 'notIn'
+  | 'gt'
+  | 'gte'
+  | 'lt'
+  | 'lte';
+
+export type DynamicValue = 'startOfYear';
+
+export interface FilterConfig {
+  field: string;
+  operator: Operator;
+  value?: string | number | boolean | Array<string | number | boolean>;
+  dynamicValue?: DynamicValue;
+}
+
+export interface AggregationConfig {
+  type: AggregationType;
+  parentField: string;
+  childField?: string;
+  currencyField?: string;
+  filters?: FilterConfig[];
+}
+
+export interface RollupDefinition {
+  parentObject: string;
+  childObject: string;
+  relationField: string;
+  childFilters?: FilterConfig[];
+  aggregations: AggregationConfig[];
+}
+
+export type RollupConfig = RollupDefinition[];
+
+export type ChildRecord = Record<string, unknown>;
+
+export interface ExecutionSummaryItem {
+  parentObject: string;
+  processed: number;
+  updated: number;
+  mode: 'full-rebuild' | 'targeted';
+  relationField: string;
+  skipped?: string;
+}