diff --git a/docs/architecture-decisions/063-integration-definitions-as-data.md b/docs/architecture-decisions/063-integration-definitions-as-data.md index faef215eb..357025bd9 100644 --- a/docs/architecture-decisions/063-integration-definitions-as-data.md +++ b/docs/architecture-decisions/063-integration-definitions-as-data.md @@ -56,6 +56,9 @@ Frigg supports **two authoring forms behind one runtime contract**: Code is the escape hatch and data is the composition. A data definition never contains code. It names building blocks that exist as code, and the app has registered and versioned them. +> §1–§6 are the proposal as first written. Where they differ from what was built (step and expression vocabulary, +> the document's shape, `name` vs `id`, routes), §7 onward is what runs. + ### 1. What is data and what stays code | Concern | Data-defined | Stays code (referenced by name + version) | @@ -421,6 +424,26 @@ does not exist at runtime, so "disabled but still firing" cannot happen. Shape B is what actually makes §4's "constant infrastructure" true. Shape A still grows the stack per definition id, just not per version. +**Neither shape is visible from outside.** Rule: nobody can tell how an integration was made, not a user, an API +caller or a log reader. Shape B's class is a *provider of integration types* (`static IntegrationProvider`, +`integrations/integration-types.js`): wherever Frigg turns `config.type` into a class (the integration use cases, +the queue worker, `GET /api/integrations` options), a published definition id resolves to a generated subclass named +by that id, with the definition's label and modules. So an integration is installed with `{ type: 'phone-crm' }`, +exactly as code is, and its logs, telemetry, schedule job ids and webhook URLs carry `phone-crm`. Registered code +classes win over a definition with the same name (logged once per process as `frigg.integrations.definition.shadowed_by_code`). Only the deployment is shared: `IntegrationBase.getDeploymentName()` +is the host's name for a generated class, so it queues and books one-time jobs on the host's queue, and its +triggers run on the host's tick (§15a); the host's webhook function +also serves `/api/{integrationType}/webhooks[/{integrationId}]` (HTTP API cannot match part of a path segment, and +literal routes of registered integrations win); app-level webhooks, one URL per vendor app, use the host's URL and fan +out to installations of every type it provides. + +**Code uses the same runtime.** `extendIntegrationClass(CodeClass, { definition, manifests })` is the third shape: a +hand-written class that embeds a definition keeps its name, modules and handlers and gains workflows (with pauses), +syncs, UI surfaces, the generated form and run history. Its code steps may call the class's own methods (`#method`), +its own queued events are recorded as runs beside its workflows, and whatever the definition does not declare +(webhooks, the form, schedules) stays the code's. So the capabilities that once belonged to data-defined integrations +belong to integrations; how one was written is not something its users, API callers or logs can tell. `{ type: 'data-defined', definition: { id } }` still installs. + ### 14. Runtime: how an event loads and runs **Cold start (unchanged).** `loadAppDefinition()` registers the generated classes like any others. No definition is @@ -510,25 +533,36 @@ claimed once (§15a). Steps such as | `GET …/actions`, `POST …/actions/:id/options` | `GET_USER_ACTIONS`, `GET_USER_ACTION_OPTIONS` | Button workflows and backfills; a button's inputs are its options form | | `POST …/actions/:id` | any registered event | Runs a button workflow (`WF_*`), a backfill (`SYNC_*_BACKFILL`) or renders a UI surface (`UI_*`, body `{ record }`) | -Data-defined apps add a few routes of their own (`router.js`, mounted by `createIntegrationRouter` only when a -registered class is data-defined). These are the only new routes, and every one is authenticated: +Apps that register the runtime add a few routes of their own (`router.js`, mounted by `createIntegrationRouter` when +a registered class carries `DataDefined`). Every one is authenticated, and the installation routes answer for every +integration however it was made: one with no run history, syncs or other versions gets `[]`, or a plan with nothing +to move to, never a 404 for being code. | Route | Who | What | |---|---|---| -| `GET /api/data-defined/definitions` | signed-in user | The install catalogue: the latest published version of each definition | +| `GET /api/data-defined/definitions` | signed-in user | The published ones with their counts (`GET /api/integrations` already lists them as options beside code) | | `GET /api/integrations/:id/upgrade?to=` | signed-in user, own installation | The upgrade plan (`planUpgrade`); changes nothing | | `POST /api/integrations/:id/upgrade` `{ to? }` | signed-in user, own installation | Applies the plan: uninstall steps of the old version, then install steps of the new one if it asks for a reinstall, then the new config and a reload. Answers 409 with the plan when a person must supply something first | | `GET /api/integrations/:id/runs` | signed-in user, own installation | `WorkflowRun` history, newest first (`workflowId`, `status`, `before`, `limit`) | -| `POST /admin/integrations/definitions` `{ document, notes?, publish?, strict? }` | `x-frigg-admin-api-key` | Validate and publish (or save a draft). 201 created, 200 for an identical republish, 400 invalid, 409 refused | -| `GET /admin/integrations/definitions/:id` | admin key | Every version, without documents | -| `POST /admin/integrations/definitions/:id/versions/:version/deprecate` | admin key | `PUBLISHED` → `DEPRECATED` | +| `GET /api/integrations/:id/syncs` | signed-in user, own installation | Each sync's backfill progress and poll marks (§15b) | +| `POST /api/integrations/:id/runs/:runId/decision` `{ decision, note? }` | the run's approvers, or the owner when it names none | Decide a paused approval (§15c). Answers 202 and carries the run on through the queue. An approver need not own the installation; anyone else gets a 404 | +| `POST /api/integration-definitions` `{ document, notes?, publish? }` | signed-in user | Publish (or draft) an integration of the user's own workspace: listed and installable there, invisible elsewhere. 409 `ID_TAKEN` for an id owned by the app's catalogue or a code integration (both public); an id owned by another workspace answers 404 `NOT_FOUND`, identical to reading it, so ids cannot be probed (ADR-052 §3) | +| `GET /api/integration-definitions/:id` | signed-in user, own workspace | Every version (drafts too) and the newest document, for the editor; 404 for anyone else | +| `POST /admin/integrations/versions` `{ document, notes?, publish?, strict? }` | `x-frigg-admin-api-key` | Validate and publish (or save a draft). 201 created, 200 for an identical republish, 400 invalid, 409 refused | +| `GET /admin/integrations/:id/versions` | admin key | Every version, without documents | +| `POST /admin/integrations/:id/versions/:version/deprecate` | admin key | `PUBLISHED` → `DEPRECATED` | The admin routes live in core's existing `/admin/*` namespace, beside `/admin/scripts` and `/admin/db-migrate`, behind the same -`validateAdminApiKey` middleware (`handlers/middleware/admin-auth`). They are deliberately not under `/api/admin`, which would be a -third namespace. The user routes stay where they are until the `/api/v2` route registry (PR #671) lands; they move then. +`validateAdminApiKey` middleware (`handlers/middleware/admin-auth`); `/api/admin` would have been a third namespace. The user routes +stay where they are until the `/api/v2` route registry (PR #671) lands. Every route the router serves is listed once, with method, +path and auth, in the exported `DEFINITION_ROUTES`, so that move is a data change. + +**Follow-up: these routes move to `/api/v2`.** The paths above are v0.1's. [ADR-053](./053-management-api-versioning.md) puts every new Management API resource under `/api/v2/`, and the route registry in #671 is how they get there. Moving them is a separate change, and the paths in this table will change with it. + +**What "workspace" means in this table.** The workspace routes (`/api/integration-definitions`) scope a definition to the authenticated user: the workspace id is `user.getId()`. That is a stand-in, not an ADR-056 owner and organization. Aligning it with ADR-056 (owner plus `organizationId`, `visible(context)`, binding scope) is the follow-up in §5. -Installing needs no new route. `POST /api/integrations` with `config.definition: { id }` pins the latest published -version, and its checksum, in `onCreate`. An app that bundles its definitions (`memoryDefinitions`) has no store to +Installing needs no new route. `POST /api/integrations` with `config: { type: '' }` pins the latest +published version, and its checksum, in `onCreate`. An app that bundles its definitions (`memoryDefinitions`) has no store to publish to, so the admin routes answer 501. A host renderer (the HubSpot card) therefore calls Frigg through an authenticated path. Proving the host request is @@ -574,6 +608,11 @@ or on a polling sync (`ongoing: { "mode": "poll", "every", "configurable" }`). ` trigger entries (`compiled.triggers`), and the data-defined class resolves them per installation: it declares `Definition.instanceTriggers: { tick: '5m' }`, and its static `triggersFor({ config })` compiles the installation's pinned definition with its config at each tick. Each entry's handler runs its compiled event. +The shared class's tick also covers the integration types it provides (`static triggerTypes()`): a published +definition's installations carry their own `config.type`, so the tick pages each type in turn, and a continuation +cursor names the type it stopped in. A code class that embeds a definition (`extendIntegrationClass`) keeps its own +`Definition.triggers` beside the compiled ones, and declares `instanceTriggers` only when its definition has +schedule-triggered workflows or polling syncs. | Layer | What | File | |---|---|---| diff --git a/packages/core/handlers/integration-event-dispatcher.js b/packages/core/handlers/integration-event-dispatcher.js index 5501a4eca..1119630e9 100644 --- a/packages/core/handlers/integration-event-dispatcher.js +++ b/packages/core/handlers/integration-event-dispatcher.js @@ -19,9 +19,13 @@ class IntegrationEventDispatcher { } async dispatchJob({ event, data, context, delivery }) { - return this._dispatch(event, (instance, handler) => - handler.call(instance, { data, context, delivery }) - ); + return this._dispatch(event, (instance, handler) => { + const invoke = () => handler.call(instance, { data, context, delivery }); + // An integration that keeps run history records each queued event as a run (data-defined runtime). + return typeof instance.runEvent === 'function' + ? instance.runEvent(event, invoke) + : invoke(); + }); } /** diff --git a/packages/core/handlers/routers/integration-webhook-routers.js b/packages/core/handlers/routers/integration-webhook-routers.js index 868dc3f30..a11b8a6dd 100644 --- a/packages/core/handlers/routers/integration-webhook-routers.js +++ b/packages/core/handlers/routers/integration-webhook-routers.js @@ -24,6 +24,17 @@ for (const IntegrationClass of integrationClasses) { } const router = Router(); + const receive = async (req, res, next, integrationId) => { + try { + const dispatcher = new IntegrationEventDispatcher(new IntegrationClass()); + const dispatch = () => dispatcher.dispatchHttp({ event: 'WEBHOOK_RECEIVED', req, res, next }); + // Logs only: the id comes from an unauthenticated URL, so it must not reach the telemetry context that + // drives usage attribution. + await (integrationId ? runInContext({ [LOGGER_SCOPE_KEY]: { integrationId } }, dispatch) : dispatch()); + } catch (error) { + next(error); + } + }; const basePath = `/api/${IntegrationClass.Definition.name}-integration/webhooks`; console.log( @@ -31,48 +42,25 @@ for (const IntegrationClass of integrationClasses) { ); // General webhook route (no integration ID) - router.post(basePath, async (req, res, next) => { - try { - const integrationInstance = new IntegrationClass(); - const dispatcher = new IntegrationEventDispatcher( - integrationInstance - ); - await dispatcher.dispatchHttp({ - event: 'WEBHOOK_RECEIVED', - req, - res, - next, - }); - } catch (error) { - next(error); - } - }); + router.post(basePath, (req, res, next) => receive(req, res, next)); console.log(`│ POST ${basePath}`); // Integration-specific webhook route (with integration ID) - router.post(`${basePath}/:integrationId`, async (req, res, next) => { - try { - const integrationInstance = new IntegrationClass(); - const dispatcher = new IntegrationEventDispatcher( - integrationInstance - ); - // Logs only: the id comes from an unauthenticated URL, so it must - // not reach the telemetry context that drives usage attribution. - await runInContext( - { [LOGGER_SCOPE_KEY]: { integrationId: req.params.integrationId } }, - () => - dispatcher.dispatchHttp({ - event: 'WEBHOOK_RECEIVED', - req, - res, - next, - }) - ); - } catch (error) { - next(error); - } - }); + router.post(`${basePath}/:integrationId`, (req, res, next) => receive(req, res, next, req.params.integrationId)); console.log(`│ POST ${basePath}/:integrationId`); + + // A provider (the shared runtime for published integration definitions) also receives the webhooks of every + // type it provides, at each type's own URL. Receipt has no database, so the type in the URL is not looked up + // here: the delivery is queued on this deployment and the worker hydrates the installation by its id. + if (IntegrationClass.IntegrationProvider) { + const providedPath = '/api/:integrationType/webhooks'; + // Registered integrations keep their own routes (a local server mounts every router on one app). + const registered = new Set(integrationClasses.map((c) => `${c.Definition.name}-integration`)); + const isProvidedPath = (req) => /^[A-Za-z0-9_.-]+-integration$/.test(req.params.integrationType || '') && !registered.has(req.params.integrationType); + router.post(providedPath, (req, res, next) => (isProvidedPath(req) ? receive(req, res, next) : next())); + router.post(`${providedPath}/:integrationId`, (req, res, next) => (isProvidedPath(req) ? receive(req, res, next, req.params.integrationId) : next())); + console.log(`│ POST /api/-integration/webhooks[/:integrationId]`); + } console.log('│'); handlers[`${IntegrationClass.Definition.name}Webhook`] = { diff --git a/packages/core/handlers/routers/integration-webhook-routers.test.js b/packages/core/handlers/routers/integration-webhook-routers.test.js index aa0150ff3..68b37db29 100644 --- a/packages/core/handlers/routers/integration-webhook-routers.test.js +++ b/packages/core/handlers/routers/integration-webhook-routers.test.js @@ -86,6 +86,20 @@ jest.mock('../app-definition-loader', () => { }; } + // A provider: the shared runtime for published integration definitions. + class ProviderIntegration extends IntegrationBase { + static Definition = { name: 'runtime-host', version: '1.0.0', modules: {}, webhooks: { enabled: true } }; + static IntegrationProvider = { resolve: async () => null, list: async () => [] }; + constructor(params) { + super(params); + this.events = { + WEBHOOK_RECEIVED: { + handler: async ({ req, res }) => res.status(200).json({ receivedBy: 'runtime-host', integrationId: req.params.integrationId ?? null, type: req.params.integrationType ?? null }), + }, + }; + } + } + return { loadAppDefinition: () => ({ integrations: [ @@ -93,6 +107,7 @@ jest.mock('../app-definition-loader', () => { AdvancedWebhookIntegration, LoggingWebhookIntegration, NoWebhookIntegration, + ProviderIntegration, ], }), }; @@ -151,6 +166,37 @@ describe('Integration Webhook Routers', () => { }); }); + describe('types a provider offers', () => { + const post = (path) => + handlers['runtime-hostWebhook'].handler( + { + version: '2.0', + routeKey: 'POST /api/{integrationType}/webhooks/{integrationId}', + rawPath: path, + rawQueryString: '', + headers: { 'content-type': 'application/json' }, + requestContext: { http: { method: 'POST', path }, requestId: 'r' }, + body: '{}', + isBase64Encoded: false, + }, + { awsRequestId: 'req-provided' } + ); + + it("receives a provided type's webhooks at that type's own URL", async () => { + const withId = await post('/api/phone-hubspot-integration/webhooks/int-5'); + expect(withId.statusCode).toBe(200); + expect(JSON.parse(withId.body)).toEqual({ receivedBy: 'runtime-host', integrationId: 'int-5', type: 'phone-hubspot-integration' }); + const appLevel = await post('/api/phone-hubspot-integration/webhooks'); + expect(JSON.parse(appLevel.body)).toMatchObject({ receivedBy: 'runtime-host', integrationId: null }); + }); + + it('leaves registered integrations and other paths alone', async () => { + expect((await post('/api/webhook-enabled-integration/webhooks/int-5')).statusCode).toBe(404); + expect((await post('/api/users/webhooks/int-5')).statusCode).toBe(404); + expect(handlers['webhook-enabledWebhook']).toBeDefined(); + }); + }); + describe('logger scope (ADR-048)', () => { const invoke = (path) => handlers['logging-webhookWebhook'].handler( diff --git a/packages/core/integrations/data-defined/README.md b/packages/core/integrations/data-defined/README.md index e9246aaee..656165f05 100644 --- a/packages/core/integrations/data-defined/README.md +++ b/packages/core/integrations/data-defined/README.md @@ -57,9 +57,19 @@ installation (Integration.config) ─┘ │ ## How it plugs into Frigg +- **One kind of integration**: every published definition is an integration type named by its id. The shared class + is a *provider* (`static IntegrationProvider`, see `../integration-types.js`): wherever Frigg turns `config.type` + into a class (create, get, update, delete, the queue worker, `GET /api/integrations` options), a published id + resolves to a class of its own, with its own name, label and modules. Registered code classes win over a + definition with the same name. What is shared is the deployment underneath: published types queue and book one-time jobs + on the provider's queue (`IntegrationBase.getDeploymentName()`), their triggers run on the provider's tick, and their webhooks reach the provider's function. + Logs, telemetry, schedule job ids and installation webhook URLs carry each integration's own name. - **Hydration**: Frigg's use cases call `initialize()` on every hydrated integration; the class loads its pinned definition there. -- **Webhooks**: install steps register `$.install.webhookUrls.` (the URL names the module). The manifest's +- **Webhooks**: install steps register `$.install.webhookUrls.` (the URL names the module). An installation's + webhooks arrive at its own type's URL (`/api/-integration/webhooks/`), which the provider's + webhook function also serves (devtools adds `/api/{integrationType}/webhooks[/{integrationId}]` to it; literal + routes of registered integrations win). App-level webhooks use the provider's URL, one per vendor app. The manifest's `webhook` block (`webhooks.js`) splits batches and reads each event's topic, id and account; `verify` must return true; handshakes are answered at receipt. App-level webhooks (`webhook.scope: "app"`, one URL per provider app) are fanned out to the installations connected to each account. @@ -71,6 +81,25 @@ installation (Integration.config) ─┘ │ `runtime.installed`, and `accounts..` — values about the connected account that the module's manifest declares under `account` (`$.install.accounts.quickbooks.realmId`). Nothing undeclared is exposed. +## Code integrations use the same runtime + +A hand-written `IntegrationBase` subclass can embed a definition and get everything a published integration has: +workflows with delay and approval pauses, syncs, UI surfaces, the generated configure form, and run history. + +```js +class PhoneReports extends IntegrationBase { static Definition = { name: 'phone-reports', modules, … }; async notifyTeam(args) { … } } +module.exports = extendIntegrationClass(PhoneReports, { definition: require('./phone-reports.definition.json'), manifests }); +``` + +- The class keeps its name, modules, routes, schedules, lifecycle and handlers; the definition's id must be its name. +- A `code` step with `ref: "#notifyTeam"` calls that method on the integration (checked at class creation; a + published definition cannot use `#` refs). +- What the definition does not declare stays the code's: with no webhook-triggered workflow or sync, the class's + own webhook handling runs; with no settings, its own configure form; its own schedules run beside the compiled ones. +- Each queued event the class handles itself is recorded as a run (`triggerType: "code"`) in the same history as its + workflows (`IntegrationEventDispatcher.dispatchJob` → `runEvent`). The history never decides a handler's outcome. +- The definition ships with the code, so installations run the deployed version; `upgrade` says so and moves nothing. + ## Publishing, installing, upgrading Definitions live in the `IntegrationDefinitionVersion` table (Prisma models in both schemas, PostgreSQL migration @@ -84,13 +113,31 @@ const DataDefined = defineSharedIntegrationClass({ definitions: createDefinition | Route | Auth | | |---|---|---| -| `POST /admin/integrations/definitions` `{ document, notes?, publish?, strict? }` | admin key | Validate (both layers) and publish, or save a draft with `publish: false` | -| `GET /admin/integrations/definitions/:id` | admin key | Version history | -| `POST /admin/integrations/definitions/:id/versions/:v/deprecate` | admin key | Stop new installs of a version | -| `GET /api/data-defined/definitions` | user | What can be installed | -| `POST /api/integrations` with `config: { type, definition: { id } }` | user | Install; pins the latest published version and its checksum | +| `POST /admin/integrations/versions` `{ document, notes?, publish?, strict? }` | admin key | Validate (both layers) and publish, or save a draft with `publish: false` | +| `GET /admin/integrations/:id/versions` | admin key | Version history | +| `POST /admin/integrations/:id/versions/:v/deprecate` | admin key | Stop new installs of a version | +| `GET /api/integrations` | user | What can be installed: code and published integrations in one list of options | +| `GET /api/data-defined/definitions` | user | The published ones, with their counts | +| `POST /api/integrations` with `config: { type: '' }` | user | Install; pins the latest published version and its checksum (`{ type: 'data-defined', definition: { id } }` still works) | | `GET` / `POST /api/integrations/:id/upgrade` `{ to? }` | user | Plan / apply an upgrade (409 with the plan while something is needed) | | `GET /api/integrations/:id/runs` | user | Run history, newest first | +| `GET /api/integrations/:id/syncs` | user | Each sync's backfill progress and poll marks | +| `POST /api/integrations/:id/runs/:runId/decision` `{ decision, note? }` | user | Decide a paused approval. The approval step's `approvers` may decide without owning the installation; anyone else gets a 404 | + +A signed-in user can also build their own: `POST /api/integration-definitions` `{ document, notes?, publish? }` publishes +into their workspace (the user's Frigg id; `IntegrationDefinitionVersion.workspaceId`). It is then listed and +installable for that workspace like any other integration (its install option carries `editable: true`, the cue for +the editor views), and invisible to everyone else: not listed, not resolvable when they install, not readable. +`GET /api/integration-definitions/:id` gives its owner every version, drafts included, and the newest document for +the editor. An id has one owner. The app's catalogue's ids and a code integration's name are public, and refused with +409 `ID_TAKEN`; an id owned by another workspace answers 404 `NOT_FOUND`, byte for byte what reading it answers, so one +tenant cannot learn that another's id exists (ADR-052 section 3). (The admin route, which is not a tenant, still gets 409.) +A published definition whose name matches a registered code integration is ignored (code wins); that is logged once per +process as `frigg.integrations.definition.shadowed_by_code`. PostgreSQL migration `20261019120000_definition_workspace`. + +The installation +routes answer for every integration, code included: one with no run history, syncs or other versions gets `[]`, or a +plan that says there is nothing to move to, never a 404 for how it was made. Published versions cannot change. Publishing a document that is identical to a published one returns that row. A different document at the same version is refused. So is a version that is not newer than the latest, and so is a diff --git a/packages/core/integrations/data-defined/data-defined-integration.js b/packages/core/integrations/data-defined/data-defined-integration.js index 0c0f3779d..8982a76ff 100644 --- a/packages/core/integrations/data-defined/data-defined-integration.js +++ b/packages/core/integrations/data-defined/data-defined-integration.js @@ -75,7 +75,43 @@ function defineSharedIntegrationClass({ name = 'data-defined', definitions, mani * entity reverse lookup, then each candidate's record, because one account can be used by integrations of other * types too, and the worker would hydrate those as their own class. */ -function friggLookup(typeName) { +/** + * A code integration that also embeds a definition: the class keeps its own name, modules, handlers and lifecycle, + * and gains what a published integration has — workflows (with delay and approval pauses), syncs, the generated + * configure form, UI surfaces, versions and run history — through the same records and routes. Code is the escape + * hatch inside it: a `code` step whose ref is `#` calls that method on the integration. + * + * class HubspotSlack extends IntegrationBase { static Definition = { name: 'hubspot-slack', modules, … }; … } + * module.exports = extendIntegrationClass(HubspotSlack, { definition: require('./hubspot-slack.definition.json'), manifests }); + * + * What the definition does not declare stays the code's: with no webhook-triggered workflow or sync the class's own + * webhook handling runs; with no settings, its own configure form; its own triggers run beside the compiled ones. + */ +function extendIntegrationClass(CodeClass, { definition, manifests, runCode, runs, lookup }) { + const name = CodeClass?.Definition?.name; + if (!name) throw new Error('extendIntegrationClass needs an IntegrationBase subclass with a static Definition.name'); + if (!definition || definition.id !== name) throw new Error(`The definition embedded in "${name}" must have the id "${name}" (it has "${definition?.id}")`); + const { validate } = require('./validate'); + const codeMethods = []; + for (let proto = CodeClass.prototype; proto && proto !== Object.prototype; proto = Object.getPrototypeOf(proto)) { + for (const k of Object.getOwnPropertyNames(proto)) if (k !== 'constructor' && typeof Object.getOwnPropertyDescriptor(proto, k).value === 'function') codeMethods.push(k); + } + const problems = (validate(definition, manifests, { codeMethods }).errors || []); + if (problems.length) throw new Error(`The definition embedded in "${name}" is not valid: ${problems.map((e) => `${e.path || ''} ${e.message}`.trim()).join('; ')}`); + return makeClass({ + name, defaultId: name, definitions: memoryDefinitions([definition]), manifests, runCode, IntegrationBase: CodeClass, lookup, runs, + modules: CodeClass.Definition.modules, display: CodeClass.Definition.display, version: CodeClass.Definition.version, embedded: true, + }); +} + +/** Whether a definition has recurring work of its own (a schedule-triggered workflow or a polling sync). */ +const hasRecurring = (def) => Boolean((def?.workflows || []).some((w) => w.trigger?.type === 'schedule') || (def?.syncs || []).some((sy) => sy.ongoing?.mode === 'poll')); +/** Whether a definition receives webhooks itself (a webhook-triggered workflow or a webhook-driven sync). */ +const ownsWebhooks = (def) => Boolean((def?.workflows || []).some((w) => w.trigger?.type === 'webhook') || (def?.syncs || []).some((sy) => sy.ongoing?.mode === 'webhook')); +/** Whether a definition generates the configure form (it declares settings, at its level or a workflow's). */ +const ownsConfigForm = (def) => Boolean(Object.keys(def?.settings || {}).length || (def?.workflows || []).some((w) => Object.keys(w.settings || {}).length)); + +function friggLookup(accepts) { return async (externalId, moduleName) => { const { createIntegrationCommands } = require('../../application/commands/integration-commands'); const commands = createIntegrationCommands(); @@ -83,36 +119,50 @@ function friggLookup(typeName) { const out = []; for (const integrationId of ids) { const found = await commands.loadIntegrationContextById(integrationId); - if (found?.context?.record?.config?.type === typeName) out.push(integrationId); + if (await accepts(found?.context?.record?.config?.type)) out.push(integrationId); } return out; }; } -function makeClass({ name, defaultId, definitions, manifests, runCode, IntegrationBase, modules, display, version, lookup, runs }) { +function makeClass({ name, defaultId, definitions, manifests, runCode, IntegrationBase, modules, display, version, lookup, runs, embedded = false }) { const Base = IntegrationBase || require('../integration-base').IntegrationBase; const id = name; - const findInstallations = lookup || friggLookup(name); // Where workflow runs are kept: the WorkflowRun table by default (loaded lazily, so tests and tools without a // database can pass their own store or `false` to keep none). let runStore = runs; + // App-level webhooks fan out to the installations of every type this deployment runs. + const findInstallations = (externalId, moduleName, Cls) => (lookup || friggLookup(async (type) => type === id || type === Cls.Definition.name || Boolean(await DataDefinedIntegration.IntegrationProvider?.resolve(type))))(externalId, moduleName); const runRepository = () => (runStore === undefined ? (runStore = require('./repositories/workflow-run-repository').createWorkflowRunRepository()) : runStore); - const publicBase = () => `${process.env.FRIGG_PUBLIC_URL || ''}/api/${id}-integration/webhooks`; + const baseFor = (typeName) => `${process.env.FRIGG_PUBLIC_URL || ''}/api/${typeName}-integration/webhooks`; const pinnedDefinitions = new Map(); // `${id}@${version}` -> definition, for the trigger tick + // A code class embedding a definition provisions the per-installation tick only when its definition has + // schedule-triggered workflows or polling syncs; its own Definition.triggers are provisioned as they are. + const embeddedRecurring = embedded && hasRecurring(definitions.latest(defaultId)); + // A code class's own Definition.triggers (IntegrationBase.triggersFor), kept beside the compiled ones. + const codeTriggers = async (Cls, installation) => (embedded && typeof Base.triggersFor === 'function' ? Base.triggersFor.call(Cls, installation) : []); class DataDefinedIntegration extends Base { - // version is display only; installations pin their own. instanceTriggers: schedule-triggered workflows and - // polling syncs are this class's recurring triggers, resolved per installation (triggersFor) on one 5m tick. - static Definition = { name, version, modules, webhooks: { enabled: true }, display, instanceTriggers: { tick: INSTANCE_TICK } }; + // version is display only; installations pin their own. A code class keeps every part of its Definition. + // instanceTriggers: schedule-triggered workflows and polling syncs are this class's recurring triggers, + // resolved per installation (triggersFor) on one 5m tick. + static Definition = embedded + ? { ...Base.Definition, webhooks: Base.Definition.webhooks ?? { enabled: true }, ...(embeddedRecurring ? { instanceTriggers: { tick: INSTANCE_TICK } } : {}) } + : { name, version, modules, webhooks: { enabled: true }, display, instanceTriggers: { tick: INSTANCE_TICK } }; /** * The recurring triggers of one installation (ADR-055): its pinned definition compiled with its config, so a * switched-off workflow has none and a configurable one carries the installation's choice. Each runs its - * compiled event. Read by the shared tick from the record alone (no modules, no credentials). + * compiled event. A code class embedding a definition keeps its own Definition.triggers beside them. Read by + * the shared tick from the record alone (no modules, no credentials). */ - static async triggersFor({ config = {} } = {}) { - const pin = config.definition || (defaultId ? { id: defaultId } : null); - if (!pin?.id) return []; + static async triggersFor(installation = {}) { + const config = installation.config || {}; + const own = await codeTriggers(this, installation); + const fixedId = this.definitionId; + let pin = config.definition || (fixedId ? { id: fixedId } : null); + if (!pin?.id) return own; + if (embedded) pin = { id: pin.id }; // the definition ships with the code: the deployed version runs let def; if (pin.version) { // Published versions are immutable, so a tick over many installations reads each one once. @@ -125,12 +175,27 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati } else { def = await definitions.latest(pin.id); } - return def ? triggersOf(def, config) : []; + return def ? [...own, ...triggersOf(def, config)] : own; + } + + /** + * The installation types one tick of this class covers (RunTriggerTick): its own, plus, for the shared class, + * every published definition it provides, because each of those is an integration type of its own that runs + * on this deployment. + */ + static async triggerTypes() { + const own = this.Definition.name; + if (!this.IntegrationProvider) return [own]; + const provided = (await this.IntegrationProvider.list()).map((c) => c.Definition.name); + return [...new Set([own, ...provided])]; } /** How the routers find data-defined classes and the store and manifests they run on. */ static DataDefined = { definitions, manifests, shared: !defaultId, runs: () => runRepository() }; + /** Shape A: the one definition this class runs. A class generated for a published definition sets its own. */ + static definitionId = defaultId; + /** * Frigg's use cases construct an integration from its record and call initialize() before anything else * (create, update, delete, webhooks, user actions, config options). So this is where the pinned definition is @@ -138,19 +203,24 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati * before its installation is known) has no pin and loads nothing. */ async initialize() { - if (this.config?.definition || defaultId) await this.loadDefinition(); + if (this.config?.definition || this.constructor.definitionId) await this.loadDefinition(); return super.initialize(); } async loadDefinition() { - let pin = this.config?.definition || (defaultId ? { id: defaultId } : null); + const fixedId = this.constructor.definitionId; + let pin = this.config?.definition || (fixedId ? { id: fixedId } : null); + if (fixedId && pin.id !== fixedId) throw new Error(`This installation pins "${pin.id}", but its type is "${fixedId}"`); if (!pin?.id) throw new Error('This installation does not name a definition (config.definition.id)'); + // A definition embedded in code ships with the code: an installation runs the version deployed now. + if (embedded) pin = { id: pin.id }; if (!pin.version) { // Installed by id alone: pin the latest published version now (onCreate persists the pin). const latest = await definitions.latest(pin.id); if (!latest) throw new Error(`No published version of "${pin.id}"`); pin = { id: pin.id, version: latest.version, checksum: await definitions.checksumOf?.(pin.id, latest.version) || undefined }; - if (this.config?.definition) this.config = { ...this.config, definition: pin }; + // Persisted by onCreate, so this installation stays on this version until it is upgraded. + this.config = { ...(this.config || {}), definition: pin }; } this.def = await definitions.get(pin.id, pin.version, { checksum: pin.checksum }); if (!this.def) throw new Error(`Definition ${pin.id}@${pin.version} is not published`); @@ -169,7 +239,18 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati /** Ports for a definition: Api calls through the manifest, and links/dedupe over IntegrationMapping. */ portsFor(def) { return makePorts({ - def, manifests, runCode, + def, manifests, + // `#method` is a method of this integration (a code class embedding a definition); anything else is a + // registered code step. + runCode: (ref, args) => { + if (typeof ref === 'string' && ref.startsWith('#')) { + const fn = this[ref.slice(1)]; + if (typeof fn !== 'function') throw new Error(`code: this integration has no method "${ref.slice(1)}"`); + return fn.call(this, args); + } + if (!runCode) throw new Error('code: this runtime has no code steps registered'); + return runCode(ref, args); + }, // The shared class registers modules by module name; a definition may name them by its own keys. callApi: (moduleKey, method, args) => this.moduleFor(moduleKey).api[method](args), store: this.mappingStore(), @@ -215,6 +296,7 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati * X-Hook-Secret), otherwise queue the delivery as Frigg always does. The URL names the module (?module=). */ async onWebhookReceived({ req, res }) { + if (embedded && !ownsWebhooks(this.def)) return super.onWebhookReceived({ req, res }); const manifest = manifests[req.query?.module]; const h = manifest && handshakeFor(manifest, { body: req.body, headers: req.headers, query: req.query }); if (h) { @@ -229,10 +311,13 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati * connected to each account the delivery is about and queue it for each. With one: verify it, split it into * module events, and queue one message per enabled workflow listening for each event. */ - async onWebhook({ data = {} }) { + async onWebhook(params) { + if (embedded && !ownsWebhooks(this.def)) return super.onWebhook(params); + const { data = {} } = params || {}; if (data.module && data.topic) return this.routeEvents(data.module, [{ topic: data.topic, event: data.body, eventId: data.eventId || data.body?.id || null }]); const delivery = { body: data.body, headers: data.headers || {}, query: data.query || {} }; - if (!this.def) return this.fanOutAppWebhook(delivery); + // No installation yet (an app-level webhook): find the ones it is for. + if (!this.id || !this.def) return this.fanOutAppWebhook(delivery); const keys = modulesFor(this.def, manifests, delivery); if (keys.length !== 1) { this.logWebhook('webhook.unrouted', { module: delivery.query.module || null, candidates: keys }); return { routed: 0 }; } @@ -266,7 +351,7 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati const accounts = new Set(eventsIn(manifest, delivery).map((e) => e.account).filter(Boolean)); let queued = 0; for (const account of accounts) { - for (const integrationId of await findInstallations(account, moduleName)) { + for (const integrationId of await findInstallations(account, moduleName, this.constructor)) { await this.queueWebhook({ integrationId, body: delivery.body, headers: delivery.headers, query: delivery.query }); queued += 1; } @@ -290,7 +375,7 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati // Every onInstall step with an id keeps its result, so onUninstall can undo exactly what install did // ($.install.runtime.installed.). Saved even when empty, so a re-install replaces stale ids. const installed = Object.fromEntries((this.def.lifecycle?.onInstall || []).filter((st) => st.id).map((st) => [st.id, out.steps?.[st.id] ?? null])); - await this.patchConfig({ ...(this.config?.definition ? { definition: this.config.definition } : {}), runtime: { ...(this.config?.runtime || {}), installed } }); + await this.patchConfig({ definition: this.config.definition, runtime: { ...(this.config?.runtime || {}), installed } }); const result = await super.onCreate(params); // Syncs that copy what already exists at install time start their backfill now (through the queue). for (const sy of this.def.syncs || []) { @@ -310,18 +395,48 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati return super.onDelete(params); } - /** This installation's triggers, from the definition it already loaded and its current config. */ + /** + * This installation's triggers, from the definition it already loaded and its current config, beside a code + * class's own Definition.triggers. + */ async triggerDeclarations() { - return this.def ? triggersOf(this.def, this.config || {}) : this.constructor.triggersFor({ config: this.config || {} }); + if (!this.def) return this.constructor.triggersFor({ config: this.config || {}, record: this.record }); + const own = await codeTriggers(this.constructor, { config: this.config || {}, record: this.record }); + return [...own, ...triggersOf(this.def, this.config || {})]; + } + + /** + * The queue worker's seam (IntegrationEventDispatcher.dispatchJob): a code integration's own queued events are + * runs too, recorded in the same history as its workflows. Compiled events record their own runs (with steps). + * As with workflows, the history never decides the outcome: a failed write is logged and the handler's result + * or error stands. + */ + async runEvent(event, invoke) { + if (!embedded || (this.compiledEventNames || []).includes(event)) return invoke(); + const startedAt = new Date().toISOString(); + const write = async (status, error) => { + try { + const store = runRepository(); + if (store) await store.createRun({ integrationId: this.id, definitionId: name, definitionVersion: this.def?.version, workflowId: event, triggerType: 'code', status, error: error ? String(error.message || error) : null, retryable: Boolean(error), trace: [], startedAt, endedAt: new Date().toISOString() }); + } catch (e) { + try { require('../../logs').getLogger('frigg.data-defined').warn('Workflow run not recorded', { eventName: 'frigg.data-defined.run.not_recorded', integrationId: this.id, workflowId: event, cause: e?.message }); } catch { /* ignore */ } + } + }; + let out; + try { out = await invoke(); } catch (e) { await write('failed', e); throw e; } + await write('completed'); + return out; } /** The configure form (GET /api/integrations/:id/config/options), generated from the definition. */ async getConfigOptions() { + if (embedded && !ownsConfigForm(this.def)) return super.getConfigOptions(); return buildConfigForm(this.def, this.config || {}, (path, values, integrationSettings) => this.optionsFor(path, values, integrationSettings)); } /** The same form re-filled from the values chosen so far (POST …/config/options/refresh): dependent dropdowns. */ async refreshConfigOptions(params = {}) { + if (embedded && !ownsConfigForm(this.def)) return super.refreshConfigOptions(params); const values = params.config || params.formData || params.data || params; return buildConfigForm(this.def, { ...(this.config || {}), ...values }, (path, v, integrationSettings) => this.optionsFor(path, v, integrationSettings)); } @@ -332,6 +447,7 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati /** Required settings come from the definition; checked without calling any API (the base class renders the form). */ async validateConfig() { + if (embedded && !ownsConfigForm(this.def)) return super.validateConfig(); return missingRequired(this.def, this.config || {}).length > 0; } @@ -346,9 +462,12 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati } installation() { - const base = publicBase(); + // An installation's webhooks arrive at its own type's URL. App-level webhooks (one URL registered with the + // vendor for every account) arrive at the deployment's URL, which finds the installations they are for. + const base = baseFor(this.constructor.Definition.name); + const appBase = baseFor(this.constructor.getDeploymentName?.() || id); const webhookUrls = Object.fromEntries(Object.entries(this.def?.modules || {}).map(([key, m]) => [key, manifests[m.module]?.webhook?.scope === 'app' - ? `${base}?module=${encodeURIComponent(m.module)}` + ? `${appBase}?module=${encodeURIComponent(m.module)}` : `${base}/${this.id}?module=${encodeURIComponent(m.module)}`])); return { ...(this.config || {}), integrationId: this.id, accounts: this.accounts(), webhookUrl: `${base}/${this.id}`, webhookUrls }; } @@ -370,6 +489,8 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati * install steps changed, undoes the old ones before running the new ones (it never leaves the old webhooks). */ async upgradeTo(toVersion, { apply = false } = {}) { + // An embedded definition's version is the code's: a deploy moves every installation, nothing else does. + if (embedded) return { allowed: false, applied: false, needs: [], actions: [], from: this.def?.version ?? null, to: this.def?.version ?? null, reason: 'This integration moves to a new version when its code is deployed' }; const { PlanInstallationUpgrade } = require('./use-cases/definition-use-cases'); const plan = await new PlanInstallationUpgrade({ catalog: definitions }).execute({ config: this.config, toVersion }); if (!apply || !plan.allowed) return { ...plan, applied: false }; @@ -475,11 +596,13 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati * An approval decision (POST /api/integrations/:id/runs/:runId/decision). Checked here, carried out through * the queue (the steps after an approval may take longer than an HTTP request should). */ - async decide(runId, { decision, userId, note } = {}) { + async decide(runId, { decision, userId, note, isOwner = true } = {}) { const { DefinitionError } = require('./use-cases/definition-use-cases'); - if (!['approve', 'reject'].includes(decision)) throw new DefinitionError('decision is approve or reject', { status: 400, code: 'BAD_DECISION' }); const store = this.mappingStore(); const cp = await store.get(`wait:${runId}`); + // Someone who neither owns the installation nor is named on this run learns nothing about it. + if (!isOwner && !(cp?.wait?.approvers || []).includes(String(userId))) throw new DefinitionError(`Integration ${this.id} not found`, { status: 404, code: 'NOT_FOUND' }); + if (!['approve', 'reject'].includes(decision)) throw new DefinitionError('decision is approve or reject', { status: 400, code: 'BAD_DECISION' }); if (!cp) throw new DefinitionError(`Run ${runId} is not waiting`, { status: 404, code: 'NOT_WAITING' }); if (cp.wait.kind !== 'approval') throw new DefinitionError(`Run ${runId} is waiting on a ${cp.wait.kind}, not an approval`, { status: 409, code: 'NOT_AN_APPROVAL' }); if (cp.wait.approvers?.length && !cp.wait.approvers.includes(String(userId))) throw new DefinitionError('You are not an approver of this run', { status: 403, code: 'NOT_AN_APPROVER' }); @@ -556,9 +679,70 @@ function makeClass({ name, defaultId, definitions, manifests, runCode, Integrati } } Object.defineProperty(DataDefinedIntegration, 'name', { value: `DataDefined_${id.replace(/-/g, '_')}` }); + if (!defaultId) DataDefinedIntegration.IntegrationProvider = definitionProvider(DataDefinedIntegration, definitions); return DataDefinedIntegration; } +/** + * The shared class as a provider of integration types (integration-types.js): every published definition is an + * integration type named by its id, run by a class of its own. The class differs from code only in where it gets its + * behaviour; it runs on the shared deployment (queue, functions) of the class that provides it. + */ +function definitionProvider(Host, definitions) { + const cache = new Map(); // id -> { version, cls } + const classFor = (entry) => { + const hit = cache.get(entry.id); + if (hit && hit.version === entry.version && hit.workspaceId === (entry.workspaceId ?? null)) return hit.cls; + const hostModules = Host.Definition.modules; + const modules = {}; + for (const [key, m] of Object.entries(entry.modules || {})) { + const moduleName = typeof m === 'string' ? m : m?.module; + if (!hostModules[moduleName]) return null; // an app module this deployment does not have: not installable here + modules[key] = hostModules[moduleName]; + } + const display = entry.display || {}; + class PublishedIntegration extends Host { + static Definition = { name: entry.id, version: entry.version, modules, webhooks: { enabled: true }, display: { label: display.label, description: display.description, category: display.category } }; + static definitionId = entry.id; + static Host = Host; + /** The workspace that built it (null: the app's own catalogue). */ + static workspaceId = entry.workspaceId ?? null; + + /** The install option (GET /api/integrations): the base class reports its first module's name as the type, + * which is not this integration's name. */ + static getOptionDetails() { + return { + type: entry.id, hasUserConfig: true, isMany: false, requiresNewEntity: false, + // Built by this workspace: its page shows the editor views (rule: otherwise identical). + ...(entry.workspaceId ? { editable: true } : {}), + display: { name: display.label, description: display.description, detailsUrl: display.detailsUrl, icon: display.icon }, + }; + } + } + Object.defineProperty(PublishedIntegration, 'IntegrationProvider', { value: undefined }); + Object.defineProperty(PublishedIntegration, 'name', { value: `Integration_${entry.id.replace(/[^0-9A-Za-z]/g, '_')}` }); + cache.set(entry.id, { version: entry.version, workspaceId: entry.workspaceId ?? null, cls: PublishedIntegration }); + return PublishedIntegration; + }; + const visible = (cls, context) => Boolean(cls) && (!context?.workspaceId || cls.workspaceId == null || cls.workspaceId === String(context.workspaceId)); + return { + async resolve(type, context) { + if (!type || type === Host.Definition.name) return null; + let cls = cache.get(type)?.cls; + if (!cls) { + const latest = await definitions.latest(type); + if (!latest) return null; + cls = classFor({ ...latest, workspaceId: (await definitions.ownerOf?.(type)) ?? null }); + } + return visible(cls, context) ? cls : null; + }, + async list(context) { + const entries = (await definitions.list?.(context)) || []; + return entries.map(classFor).filter((cls) => visible(cls, context)); + }, + }; +} + /** In-memory definition store for tests and local dev. Production: the IntegrationDefinitionVersion table. */ function memoryDefinitions(docs) { const byKey = new Map(docs.map((d) => [`${d.id}@${d.version}`, d])); @@ -575,4 +759,4 @@ function memoryDefinitions(docs) { }; } -module.exports = { defineIntegrationClass, defineSharedIntegrationClass, memoryDefinitions }; +module.exports = { defineIntegrationClass, defineSharedIntegrationClass, extendIntegrationClass, memoryDefinitions }; diff --git a/packages/core/integrations/data-defined/index.js b/packages/core/integrations/data-defined/index.js index 930f29cdf..fdfca851c 100644 --- a/packages/core/integrations/data-defined/index.js +++ b/packages/core/integrations/data-defined/index.js @@ -9,7 +9,7 @@ const { compile, routeWebhook, eventName } = require('./compile'); const { execute } = require('./interpreter'); const { makePorts, runWorkflow, runQuery, renderSurface, resolveView, runLifecycle, resolveOptions } = require('./runtime'); const { diff, checkBump, planUpgrade } = require('./versioning'); -const { defineIntegrationClass, defineSharedIntegrationClass, memoryDefinitions } = require('./data-defined-integration'); +const { defineIntegrationClass, defineSharedIntegrationClass, extendIntegrationClass, memoryDefinitions } = require('./data-defined-integration'); const { BUILTINS } = require('./steps'); const { TRANSFORMS, applyMapping } = require('./transforms'); const { FILTERS } = require('./expressions'); @@ -32,7 +32,7 @@ module.exports = { validate, validateManifest, checkDefinition, checkManifest, checkInstallation, compile, routeWebhook, eventName, execute, makePorts, runWorkflow, runQuery, renderSurface, resolveView, runLifecycle, diff, checkBump, planUpgrade, - defineIntegrationClass, defineSharedIntegrationClass, memoryDefinitions, + defineIntegrationClass, defineSharedIntegrationClass, extendIntegrationClass, memoryDefinitions, BUILTINS, TRANSFORMS, FILTERS, applyMapping, resolveOptions, loadManifests, loadExamples, ...useCases, diff --git a/packages/core/integrations/data-defined/repositories/definition-version-repository.js b/packages/core/integrations/data-defined/repositories/definition-version-repository.js index 19ad3e97c..4c98524ae 100644 --- a/packages/core/integrations/data-defined/repositories/definition-version-repository.js +++ b/packages/core/integrations/data-defined/repositories/definition-version-repository.js @@ -7,7 +7,7 @@ */ const { prisma } = require('../../../database/prisma'); -const FIELDS = ['definitionId', 'version', 'schemaVersion', 'status', 'checksum', 'document', 'changeLevel', 'previousVersion', 'notes', 'createdBy', 'publishedAt']; +const FIELDS = ['definitionId', 'version', 'schemaVersion', 'status', 'checksum', 'document', 'changeLevel', 'previousVersion', 'notes', 'createdBy', 'workspaceId', 'publishedAt']; const pick = (o) => Object.fromEntries(FIELDS.filter((k) => o[k] !== undefined).map((k) => [k, o[k]])); class DefinitionVersionRepositoryMongo { @@ -27,7 +27,7 @@ class DefinitionVersionRepositoryMongo { async listByStatus(statuses) { const rows = await this.prisma.integrationDefinitionVersion.findMany({ where: { status: { in: statuses } }, - select: { id: true, definitionId: true, version: true, schemaVersion: true, status: true, checksum: true, changeLevel: true, publishedAt: true, createdAt: true }, + select: { id: true, definitionId: true, version: true, schemaVersion: true, status: true, checksum: true, changeLevel: true, workspaceId: true, publishedAt: true, createdAt: true }, }); return rows.map((r) => this.toRow(r)); } diff --git a/packages/core/integrations/data-defined/router.js b/packages/core/integrations/data-defined/router.js index 261996838..c0b2eb3f5 100644 --- a/packages/core/integrations/data-defined/router.js +++ b/packages/core/integrations/data-defined/router.js @@ -1,26 +1,64 @@ /** - * HTTP routes for data-defined integrations, mounted by createIntegrationRouter when the app registers a - * data-defined class. Thin adapters: authenticate, call a use case, map the result to a status. + * HTTP routes for integration versions, run history, sync progress and approvals, mounted by createIntegrationRouter + * when the app registers the integration runtime (a class with `DataDefined`). They answer for EVERY integration, + * however it was made: an integration with no run history, syncs or other versions gets an empty answer, not a 404. + * Thin adapters: authenticate, call a use case, map the result to a status. * * Signed-in users (Frigg's usual authentication: bearer token, x-frigg headers or adopter JWT): * GET /api/data-defined/definitions what can be installed: latest published version of each + * (GET /api/integrations lists the same types as options) * GET /api/integrations/:integrationId/upgrade?to= the upgrade plan for an installation (changes nothing) * POST /api/integrations/:integrationId/upgrade { to? } apply it (refused while anything is needed) * GET /api/integrations/:integrationId/runs workflow run history, newest first (?workflowId&status&before&limit) * GET /api/integrations/:integrationId/syncs each sync's backfill progress and poll marks * POST /api/integrations/:integrationId/runs/:runId/decision { decision: approve|reject, note? } on a paused approval * + * A workspace's own integrations (signed-in user; the workspace is the user's Frigg id): + * POST /api/integration-definitions { document, notes?, publish? = true } publish (or draft) a version + * GET /api/integration-definitions/:id its versions and the latest document, for the editor + * * App admins (x-frigg-admin-api-key, the validateAdminApiKey middleware that /admin/scripts and /admin/db-migrate use): - * POST /admin/integrations/definitions { document, notes?, publish? = true, strict? } validate + publish - * GET /admin/integrations/definitions/:id every version of one definition - * POST /admin/integrations/definitions/:id/versions/:version/deprecate + * POST /admin/integrations/versions { document, notes?, publish? = true, strict? } validate + publish + * GET /admin/integrations/:id/versions every version of one integration + * POST /admin/integrations/:id/versions/:version/deprecate * - * Installing needs no new route: POST /api/integrations with config { type: '', definition: { id } } - * pins the latest published version. + * Installing needs no new route: POST /api/integrations with config { type: '' } pins the latest + * published version, exactly as a code integration is installed by its name. */ const catchAsyncError = require('express-async-handler'); const { validateAdminApiKey } = require('../../handlers/middleware/admin-auth'); -const { DefinitionError, PublishDefinitionVersion, DeprecateDefinitionVersion, GetDefinitionHistory } = require('./use-cases/definition-use-cases'); +const { DefinitionError, PublishDefinitionVersion, DeprecateDefinitionVersion, GetDefinitionHistory, GetIntegrationForDecision, GetOwnDefinition } = require('./use-cases/definition-use-cases'); + +/** An integration with one version (code that is not published as versions): nothing to move to. */ +const noUpgrade = (integration) => ({ + allowed: false, applied: false, needs: [], actions: [], + from: integration.constructor?.Definition?.version ?? null, to: integration.constructor?.Definition?.version ?? null, + reason: 'This integration has no other version to move to', +}); + +/** + * Every route this router registers, in one place: method, path and who may call it. The registrations below take their + * paths from here, so this list cannot drift from what is served. + * + * Admin routes sit in core's existing /admin/* namespace (beside /admin/scripts and /admin/db-migrate, same admin-auth + * middleware). The user routes are on their pre-/api/v2 paths and move onto the /api/v2 route registry via PR #671; having + * them listed as data is what makes that move mechanical. + * auth: 'user' = a signed-in user (the workspace is the user's Frigg id), 'admin' = x-frigg-admin-api-key. + */ +const DEFINITION_ROUTES = Object.freeze({ + catalogue: { method: 'GET', path: '/api/data-defined/definitions', auth: 'user' }, + upgradePlan: { method: 'GET', path: '/api/integrations/:integrationId/upgrade', auth: 'user' }, + upgradeApply: { method: 'POST', path: '/api/integrations/:integrationId/upgrade', auth: 'user' }, + runs: { method: 'GET', path: '/api/integrations/:integrationId/runs', auth: 'user' }, + decide: { method: 'POST', path: '/api/integrations/:integrationId/runs/:runId/decision', auth: 'user' }, + syncs: { method: 'GET', path: '/api/integrations/:integrationId/syncs', auth: 'user' }, + ownPublish: { method: 'POST', path: '/api/integration-definitions', auth: 'user' }, + ownRead: { method: 'GET', path: '/api/integration-definitions/:definitionId', auth: 'user' }, + adminPublish: { method: 'POST', path: '/admin/integrations/versions', auth: 'admin' }, + adminVersions: { method: 'GET', path: '/admin/integrations/:definitionId/versions', auth: 'admin' }, + adminDeprecate: { method: 'POST', path: '/admin/integrations/:definitionId/versions/:version/deprecate', auth: 'admin' }, +}); +const path = (key) => DEFINITION_ROUTES[key].path; function sendError(res, e) { if (e instanceof DefinitionError || e?.name === 'DefinitionError') return res.status(e.status || 400).json({ error: e.message, code: e.code, details: e.details }); @@ -41,48 +79,68 @@ function setDataDefinedRoutes(router, authenticateUser, { getIntegrationInstance const installation = async (req) => { const user = await authenticateUser.execute(req); - const integration = await getIntegrationInstance.execute(req.params.integrationId, user.getId()); - if (!integration?.constructor?.DataDefined) throw new DefinitionError('This integration is not data-defined', { status: 404, code: 'NOT_DATA_DEFINED' }); - return integration; + return getIntegrationInstance.execute(req.params.integrationId, user.getId()); }; + const runtimeOf = (integration) => integration?.constructor?.DataDefined || null; - router.route('/api/data-defined/definitions').get(guard(async (req, res) => { - await authenticateUser.execute(req); - const lists = await Promise.all(classes.map((c) => (c.DataDefined.definitions.list ? c.DataDefined.definitions.list() : []))); + router.route(path('catalogue')).get(guard(async (req, res) => { + const user = await authenticateUser.execute(req); + const lists = await Promise.all(classes.map((c) => (c.DataDefined.definitions.list ? c.DataDefined.definitions.list({ workspaceId: String(user.getId()) }) : []))); const seen = new Set(); res.json({ definitions: lists.flat().filter((d) => (seen.has(d.id) ? false : seen.add(d.id))) }); })); - router.route('/api/integrations/:integrationId/upgrade') + router.route(path('upgradePlan')) .get(guard(async (req, res) => { const integration = await installation(req); + if (typeof integration.upgradeTo !== 'function') return res.json(noUpgrade(integration)); res.json(await integration.upgradeTo(req.query.to || undefined, { apply: false })); })) .post(guard(async (req, res) => { const integration = await installation(req); + if (typeof integration.upgradeTo !== 'function') return res.status(409).json(noUpgrade(integration)); const result = await integration.upgradeTo(req.body?.to || undefined, { apply: true }); res.status(result.applied ? 200 : 409).json(result); })); - router.route('/api/integrations/:integrationId/runs').get(guard(async (req, res) => { + router.route(path('runs')).get(guard(async (req, res) => { const integration = await installation(req); - const store = integration.constructor.DataDefined.runs(); + const store = runtimeOf(integration)?.runs() || catalogClass.DataDefined.runs(); if (!store) return res.json({ runs: [] }); const { workflowId, status, before, limit } = req.query; res.json({ runs: await store.listRuns(integration.id, { workflowId, status, before, limit: limit ? Number(limit) : undefined }) }); })); - router.route('/api/integrations/:integrationId/runs/:runId/decision').post(guard(async (req, res) => { + // Approvers need not own the installation: the run names who may decide it (an approval step's approvers). + // Anyone else learns nothing, not even that the installation exists. + router.route(path('decide')).post(guard(async (req, res) => { const user = await authenticateUser.execute(req); - const integration = await getIntegrationInstance.execute(req.params.integrationId, user.getId()); - if (!integration?.constructor?.DataDefined) throw new DefinitionError('This integration is not data-defined', { status: 404, code: 'NOT_DATA_DEFINED' }); - const out = await integration.decide(req.params.runId, { decision: req.body?.decision, note: req.body?.note, userId: user.getId() }); + const { integration, isOwner } = await new GetIntegrationForDecision({ getIntegrationInstance }).execute(req.params.integrationId, user.getId()); + if (typeof integration?.decide !== 'function') throw new DefinitionError(`Run ${req.params.runId} is not waiting`, { status: 404, code: 'NOT_WAITING' }); + const out = await integration.decide(req.params.runId, { decision: req.body?.decision, note: req.body?.note, userId: user.getId(), isOwner }); res.status(202).json(out); })); - router.route('/api/integrations/:integrationId/syncs').get(guard(async (req, res) => { + router.route(path('syncs')).get(guard(async (req, res) => { const integration = await installation(req); - res.json({ syncs: await integration.syncStatus() }); + res.json({ syncs: typeof integration.syncStatus === 'function' ? await integration.syncStatus() : [] }); + })); + + // A workspace's own integrations: publish a version, read it back for the editor. The + // integration then appears to that workspace like any other; no one else can see, read or install it. + const registeredNames = new Set(integrationClasses.map((c) => c.Definition?.name).filter(Boolean)); + router.route(path('ownPublish')).post(guard(async (req, res) => { + const user = await authenticateUser.execute(req); + const { document, notes, publish } = req.body || {}; + const out = await new PublishDefinitionVersion({ repository: repository(), manifests, isReserved: (id) => registeredNames.has(id) }) + .execute({ document, notes, publish: publish !== false, workspaceId: String(user.getId()), createdBy: String(user.getId()) }); + const { document: _omit, ...version } = out.version; + res.status(out.created ? 201 : 200).json({ ...out, version }); + })); + router.route(path('ownRead')).get(guard(async (req, res) => { + const user = await authenticateUser.execute(req); + const out = await new GetOwnDefinition({ repository: repository() }).execute({ definitionId: req.params.definitionId, workspaceId: String(user.getId()) }); + res.json(out); })); const admin = (path) => router.route(path).all(validateAdminApiKey); @@ -92,16 +150,16 @@ function setDataDefinedRoutes(router, authenticateUser, { getIntegrationInstance return repo; }; - admin('/admin/integrations/definitions').post(guard(async (req, res) => { + admin(path('adminPublish')).post(guard(async (req, res) => { const { document, notes, publish, strict, createdBy } = req.body || {}; - const out = await new PublishDefinitionVersion({ repository: repository(), manifests }).execute({ document, notes, createdBy, publish: publish !== false, strict: Boolean(strict) }); + const out = await new PublishDefinitionVersion({ repository: repository(), manifests, isReserved: (id) => registeredNames.has(id) }).execute({ document, notes, createdBy, publish: publish !== false, strict: Boolean(strict) }); const { document: _omit, ...version } = out.version; res.status(out.created ? 201 : 200).json({ ...out, version }); })); - admin('/admin/integrations/definitions/:definitionId').get(guard(async (req, res) => { + admin(path('adminVersions')).get(guard(async (req, res) => { res.json({ versions: await new GetDefinitionHistory({ repository: repository() }).execute({ definitionId: req.params.definitionId }) }); })); - admin('/admin/integrations/definitions/:definitionId/versions/:version/deprecate').post(guard(async (req, res) => { + admin(path('adminDeprecate')).post(guard(async (req, res) => { const row = await new DeprecateDefinitionVersion({ repository: repository() }).execute(req.params); const { document: _omit, ...version } = row; res.json({ version }); @@ -109,4 +167,4 @@ function setDataDefinedRoutes(router, authenticateUser, { getIntegrationInstance return true; } -module.exports = { setDataDefinedRoutes }; +module.exports = { setDataDefinedRoutes, DEFINITION_ROUTES }; diff --git a/packages/core/integrations/data-defined/tests/embedded.test.js b/packages/core/integrations/data-defined/tests/embedded.test.js new file mode 100644 index 000000000..b48a5accb --- /dev/null +++ b/packages/core/integrations/data-defined/tests/embedded.test.js @@ -0,0 +1,178 @@ +/** + * A code integration embedding a definition (extendIntegrationClass): it keeps its own name, modules, handlers, + * webhooks, form and triggers, and gains workflows with approvals, code steps calling its own methods, and run + * history for its own queued events, all through the same records and routes as a published integration. + */ +const express = require('express'); +const request = require('supertest'); +const d = require('..'); +const { setDataDefinedRoutes } = require('../router'); +const { IntegrationEventDispatcher } = require('../../../handlers/integration-event-dispatcher'); +const { hubspotApi, phoneApi } = require('./helpers-fake-crm'); + +const manifests = d.loadManifests(); +const example = (id) => JSON.parse(JSON.stringify(d.loadExamples().find((x) => x.id === id))); + +class FakeBase { + constructor(p = {}) { this.events = {}; this.config = p.config; this.id = p.id; this.rows = new Map(); this.queue = []; } + async initialize() { this.registerEventHandlers(); } + registerEventHandlers() {} + async patchConfig(c) { this.config = { ...this.config, ...c }; } + async getMapping(k) { return this.rows.has(k) ? { mapping: this.rows.get(k) } : null; } + async upsertMapping(k, v) { this.rows.set(k, JSON.parse(JSON.stringify(v))); } + async createMapping(k, v) { if (this.rows.has(k)) { const e = new Error('exists'); e.name = 'MappingAlreadyExistsError'; throw e; } this.rows.set(k, v); } + async deleteMapping(k) { this.rows.delete(k); } + async queueWebhook(m) { this.queue.push(m); } + async onCreate() { this.created = true; } + async onWebhook() { return { handledBy: 'base' }; } + async getConfigOptions() { return { jsonSchema: { base: true } }; } + /** As IntegrationBase.triggersFor: Definition.triggers, each with its name. */ + static async triggersFor() { return Object.entries(this.Definition?.triggers || {}).map(([name, t]) => ({ ...t, name })); } +} + +/** The hand-written integration: its own queued event, webhook handling, form and trigger. */ +class PhoneReports extends FakeBase { + static Definition = { name: 'phone-reports', version: '2.0.0', modules: { phone: { definition: {} }, hubspot: { definition: {} } }, display: { label: 'Phone reports' }, triggers: { NIGHTLY: { every: '1d', handler: 'nightly' } } }; + constructor(p) { + super(p); + this.events = { NIGHTLY: { type: 'QUEUE', handler: async ({ data }) => this.nightly(data) }, BROKEN: { type: 'QUEUE', handler: async () => { throw new Error('boom'); } } }; + } + async nightly(data) { return { sent: true, ...data }; } + async onWebhook() { return { handledBy: 'code' }; } + async getConfigOptions() { return { jsonSchema: { code: true } }; } + /** Called by the embedded workflow as `#notifyTeam`. */ + async notifyTeam({ who }) { this.notified = (this.notified || []).concat(who); return { ok: true, who }; } +} + +/** An embedded definition: one button workflow that waits for an approval, then calls a method of the class. */ +function embeddedDefinition(over = {}) { + const def = example('phone-hubspot'); + def.id = 'phone-reports'; + def.version = '1.0.0'; + def.syncs = []; def.ui = []; def.queries = []; def.lifecycle = {}; def.settings = {}; + def.workflows = [{ id: 'escalate', label: 'Escalate', default: 'on', trigger: { type: 'userAction', label: 'Escalate' }, steps: [ + { id: 'ok', use: 'approval', with: { prompt: 'Escalate?', approvers: ['boss'] } }, + { id: 'tell', use: 'code', with: { ref: '#notifyTeam', with: { who: '$.steps.ok.by' } } }, + ] }]; + return { ...def, ...over }; +} + +function runStore() { + const rows = new Map(); + let n = 0; + return { + rows, + async createRun(r) { const row = { ...JSON.parse(JSON.stringify(r)), id: String(++n) }; rows.set(row.id, row); return row; }, + async updateRun(id, data) { const row = rows.get(String(id)); Object.assign(row, JSON.parse(JSON.stringify(data))); return row; }, + async findRun() { return null; }, + async listRuns() { return [...rows.values()]; }, + }; +} + +function setup(def = embeddedDefinition()) { + const runs = runStore(); + const Cls = d.extendIntegrationClass(PhoneReports, { definition: def, manifests, runs }); + const inst = new Cls({ id: 'int1', config: { type: 'phone-reports', settings: { fromNumber: '+15125550100' } } }); + inst.hubspot = { api: hubspotApi() }; + inst.phone = { api: phoneApi() }; + return { Cls, inst, runs }; +} + +describe('a code integration embedding a definition', () => { + it('keeps its identity: name, modules, triggers, display, and its own events beside the compiled ones', async () => { + const { Cls, inst } = setup(); + expect(Cls.Definition).toMatchObject({ name: 'phone-reports', version: '2.0.0', triggers: { NIGHTLY: { every: '1d', handler: 'nightly' } }, display: { label: 'Phone reports' } }); + expect(Cls.Definition.modules).toBe(PhoneReports.Definition.modules); + expect(Cls.getDeploymentName?.() ?? Cls.Definition.name).toBe('phone-reports'); + await inst.initialize(); + expect(Object.keys(inst.events).sort()).toEqual(['BROKEN', 'NIGHTLY', 'WF_ESCALATE']); + expect((await inst.triggerDeclarations()).map((t) => t.name)).toEqual(['NIGHTLY']); + expect(Cls.Definition.instanceTriggers).toBeUndefined(); // the embedded definition has no recurring work + expect(inst instanceof PhoneReports).toBe(true); + }); + + it('a scheduled workflow in the embedded definition is a per-installation trigger beside the code\'s own', async () => { + const def = embeddedDefinition(); + def.workflows.push({ id: 'digest', label: 'Digest', default: 'on', trigger: { type: 'schedule', every: '1h' }, steps: [{ id: 'tell', use: 'code', with: { ref: '#notifyTeam', with: { who: 'team' } } }] }); + const { Cls, inst } = setup(def); + expect(Cls.Definition).toMatchObject({ triggers: { NIGHTLY: { every: '1d' } }, instanceTriggers: { tick: '5m' } }); + expect((await Cls.triggersFor({ config: { type: 'phone-reports' } })).map((t) => t.name)).toEqual(['NIGHTLY', 'WF_DIGEST']); + await inst.initialize(); + const triggers = await inst.triggerDeclarations(); + expect(triggers.map((t) => t.name)).toEqual(['NIGHTLY', 'WF_DIGEST']); + await triggers[1].handler.call(inst, { trigger: 'WF_DIGEST', scheduledFor: '2026-10-19T10:00:00.000Z' }); + expect(inst.notified).toEqual(['team']); + // Switched off for this installation: its trigger is gone, the code's stays. + inst.config = { ...inst.config, workflows: { digest: { enabled: false } } }; + expect((await Cls.triggersFor({ config: inst.config })).map((t) => t.name)).toEqual(['NIGHTLY']); + }); + + it('a workflow pauses for its approver, then calls the class\'s own method with who decided', async () => { + const { inst, runs } = setup(); + await inst.initialize(); + const paused = await inst.events.WF_ESCALATE.handler({}); + expect(paused.status).toBe('waiting'); + await inst.decide(paused.id, { decision: 'approve', userId: 'boss', isOwner: false }); + const { event, ...data } = inst.queue.shift(); + const done = await inst.events[event].handler({ data }); + expect(done.status).toBe('completed'); + expect(inst.notified).toEqual(['boss']); + expect([...runs.rows.values()].map((r) => [r.workflowId, r.status])).toEqual([['escalate', 'completed']]); + }); + + it('its own queued events are runs in the same history; a failure is recorded and still thrown', async () => { + const { inst, runs } = setup(); + await inst.initialize(); + const out = await new IntegrationEventDispatcher(inst).dispatchJob({ event: 'NIGHTLY', data: { day: 'mon' } }); + expect(out).toEqual({ sent: true, day: 'mon' }); + await expect(new IntegrationEventDispatcher(inst).dispatchJob({ event: 'BROKEN', data: {} })).rejects.toThrow('boom'); + expect([...runs.rows.values()].map((r) => [r.workflowId, r.triggerType, r.status, r.error])).toEqual([['NIGHTLY', 'code', 'completed', null], ['BROKEN', 'code', 'failed', 'boom']]); + // A compiled workflow dispatched by the queue is one run (its own, with steps), not a second "code" run. + runs.rows.clear(); + await new IntegrationEventDispatcher(inst).dispatchJob({ event: 'WF_ESCALATE', data: {} }); + expect([...runs.rows.values()].map((r) => [r.workflowId, r.status])).toEqual([['escalate', 'waiting']]); + // A run store that fails never changes the handler's outcome. + inst.constructor.DataDefined.runs().createRun = async () => { throw new Error('db down'); }; + expect(await new IntegrationEventDispatcher(inst).dispatchJob({ event: 'NIGHTLY', data: {} })).toEqual({ sent: true }); + }); + + it('what the definition does not declare stays the code\'s: webhooks and the configure form', async () => { + const { inst } = setup(); + await inst.initialize(); + expect(await inst.onWebhook({ data: { body: {} } })).toEqual({ handledBy: 'code' }); + expect(await inst.getConfigOptions()).toEqual({ jsonSchema: { code: true } }); + // Once the definition declares settings, the form is generated from it. + const withSettings = setup(embeddedDefinition({ settings: { team: { type: 'string', label: 'Team', required: true } } })).inst; + await withSettings.initialize(); + const form = await withSettings.getConfigOptions(); + expect(JSON.stringify(form)).toContain('Team'); + }); + + it('runs the version deployed with the code, whatever an installation pinned before', async () => { + const { Cls } = setup(embeddedDefinition({ version: '1.1.0' })); + const inst = new Cls({ id: 'int1', config: { type: 'phone-reports', definition: { id: 'phone-reports', version: '1.0.0' } } }); + await inst.initialize(); + expect(inst.def.version).toBe('1.1.0'); + expect(inst.config.definition.version).toBe('1.1.0'); + }); + + it('refuses a definition with another id, an invalid one, or a #method the class does not have; a published definition cannot call #methods', () => { + expect(() => d.extendIntegrationClass(PhoneReports, { definition: { ...embeddedDefinition(), id: 'other' }, manifests })).toThrow('must have the id "phone-reports"'); + const bad = embeddedDefinition(); + bad.workflows[0].steps[1].with.ref = '#noSuchMethod'; + expect(() => d.extendIntegrationClass(PhoneReports, { definition: bad, manifests })).toThrow('no method "noSuchMethod"'); + expect(d.validate(embeddedDefinition(), manifests).errors.map((e) => e.message).join()).toContain('only a definition embedded in one can'); + }); + + it('answers the shared routes: runs, syncs and upgrade', async () => { + const { inst } = setup(); + await inst.initialize(); + await new IntegrationEventDispatcher(inst).dispatchJob({ event: 'NIGHTLY', data: {} }); + const app = express(); + app.use(express.json()); + setDataDefinedRoutes(app, { execute: async () => ({ getId: () => 'u1' }) }, { getIntegrationInstance: { execute: async () => inst, integrationRepository: { findIntegrationById: async () => ({ userId: 'u1' }) } }, integrationClasses: [inst.constructor] }); + expect((await request(app).get('/api/integrations/int1/runs')).body.runs.map((r) => r.workflowId)).toEqual(['NIGHTLY']); + expect((await request(app).get('/api/integrations/int1/syncs')).body).toEqual({ syncs: [] }); + expect((await request(app).get('/api/integrations/int1/upgrade')).body).toMatchObject({ allowed: false }); + }); +}); diff --git a/packages/core/integrations/data-defined/tests/pause.test.js b/packages/core/integrations/data-defined/tests/pause.test.js index a94418df5..be4f31e0b 100644 --- a/packages/core/integrations/data-defined/tests/pause.test.js +++ b/packages/core/integrations/data-defined/tests/pause.test.js @@ -180,7 +180,11 @@ describe('approval', () => { const a = express(); a.use(express.json()); const authenticateUser = { execute: async () => ({ getId: () => userId }) }; - const getIntegrationInstance = { execute: async () => inst }; + // As the real use case: the record names its owner, and execute() refuses anyone else. + const getIntegrationInstance = { + integrationRepository: { findIntegrationById: async (id) => (id === 'int1' ? { id, userId: 'u1' } : null) }, + execute: async (id, uid) => { if (uid !== 'u1') { const e = new Error(`Integration ${id} does not belong to User ${uid}`); e.status = 500; throw e; } return inst; }, + }; setDataDefinedRoutes(a, authenticateUser, { getIntegrationInstance, integrationClasses: [inst.constructor] }); a.use((err, req, res, next) => res.status(err.status || 500).json({ error: err.message })); // eslint-disable-line no-unused-vars return a; @@ -239,17 +243,24 @@ describe('approval', () => { } }); - it('only a named approver may decide; a bad decision or a delay cannot be decided', async () => { + it('a named approver decides without owning the installation; anyone else learns nothing; a bad decision or a delay cannot be decided', async () => { const def = approvalDef({ approvers: ['boss'] }); def.workflows.push(workflow('wait', [{ id: 'w', use: 'delay', with: { for: '1h' } }])); const { inst } = setup([def]); await inst.initialize(); const delayed = await inst.events.WF_WAIT.handler({}); - expect((await request(app(inst, 'boss')).post(`/api/integrations/int1/runs/${delayed.id}/decision`).send({ decision: 'approve' })).status).toBe(409); + expect((await request(app(inst, 'u1')).post(`/api/integrations/int1/runs/${delayed.id}/decision`).send({ decision: 'approve' })).status).toBe(409); + expect((await request(app(inst, 'boss')).post(`/api/integrations/int1/runs/${delayed.id}/decision`).send({ decision: 'approve' })).status).toBe(404); // not named on the delay const paused = await inst.events.WF_GATE.handler({}); - expect((await request(app(inst, 'u1')).post(`/api/integrations/int1/runs/${paused.id}/decision`).send({ decision: 'approve' })).status).toBe(403); + expect((await request(app(inst, 'u1')).post(`/api/integrations/int1/runs/${paused.id}/decision`).send({ decision: 'approve' })).status).toBe(403); // the owner, but not named + const stranger = await request(app(inst, 'mallory')).post(`/api/integrations/int1/runs/${paused.id}/decision`).send({ decision: 'approve' }); + expect(stranger.status).toBe(404); + expect(stranger.body.code).toBe('NOT_FOUND'); + expect((await request(app(inst, 'mallory')).post(`/api/integrations/nope/runs/${paused.id}/decision`).send({ decision: 'approve' })).status).toBe(404); expect((await request(app(inst, 'boss')).post(`/api/integrations/int1/runs/${paused.id}/decision`).send({ decision: 'maybe' })).status).toBe(400); expect((await request(app(inst, 'boss')).post(`/api/integrations/int1/runs/${paused.id}/decision`).send({ decision: 'approve' })).status).toBe(202); + expect(inst.queue).toEqual([expect.objectContaining({ resume: paused.id, decision: 'approved', by: 'boss' })]); + expect(inst.queue).toHaveLength(1); }); }); diff --git a/packages/core/integrations/data-defined/tests/persistence.test.js b/packages/core/integrations/data-defined/tests/persistence.test.js index 9b97bcf23..9face226b 100644 --- a/packages/core/integrations/data-defined/tests/persistence.test.js +++ b/packages/core/integrations/data-defined/tests/persistence.test.js @@ -115,6 +115,20 @@ describeLegs('data-defined persistence on %s', (_, url, loadClient, DefRepo, Run expect(history[0].document).toBeUndefined(); }); + it("keeps a workspace's own integration in that workspace: stored, listed to it alone, and its id is taken", async () => { + const own = { ...example('github-linear'), id: 'acme-github-linear', version: '1.0.0' }; + const publish = (document, workspaceId) => new d.PublishDefinitionVersion({ repository: repo, manifests }).execute({ document, workspaceId }); + expect((await publish(own, 'ws-acme')).version.workspaceId).toBe('ws-acme'); + const catalog = d.createDefinitionCatalog({ repository: repo }); + expect((await catalog.list({ workspaceId: 'ws-acme' })).map((x) => x.id)).toContain('acme-github-linear'); + expect((await catalog.list({ workspaceId: 'ws-other' })).map((x) => x.id)).not.toContain('acme-github-linear'); + expect(await catalog.ownerOf('acme-github-linear')).toBe('ws-acme'); + expect(await catalog.ownerOf('phone-hubspot')).toBeNull(); + await expect(publish({ ...own, version: '1.0.1' }, 'ws-other')).rejects.toMatchObject({ status: 404, code: 'NOT_FOUND' }); // not confirmed to another tenant + await expect(publish({ ...own, version: '1.0.1' }, null)).rejects.toMatchObject({ status: 409, code: 'ID_TAKEN' }); // the admin is not a tenant + await expect(publish({ ...example('phone-hubspot'), version: '9.0.0' }, 'ws-acme')).rejects.toMatchObject({ code: 'ID_TAKEN' }); + }); + it('records workflow runs and lists them newest first, filtered and paged; old runs can be deleted', async () => { const base = { integrationId, definitionId: 'phone-hubspot', definitionVersion: '1.4.0', triggerType: 'webhook', trace: [{ id: 'contact', use: 'links.find', at: 'steps[0]', status: 'ok', ms: 3 }] }; await runs.createRun({ ...base, workflowId: 'log-call', status: 'completed', eventId: 'e1', startedAt: '2026-10-18T10:00:00Z', endedAt: '2026-10-18T10:00:01Z' }); diff --git a/packages/core/integrations/data-defined/tests/route-registry.test.js b/packages/core/integrations/data-defined/tests/route-registry.test.js new file mode 100644 index 000000000..01bd2f227 --- /dev/null +++ b/packages/core/integrations/data-defined/tests/route-registry.test.js @@ -0,0 +1,62 @@ +/** + * DEFINITION_ROUTES is the one list of what the definitions router serves. It must equal what is actually mounted, and the + * admin routes must sit in core's /admin/* namespace behind the admin key (not /api/admin). + */ +jest.mock('../../../database/config', () => ({ DB_TYPE: 'mongodb', getDatabaseType: jest.fn(() => 'mongodb'), PRISMA_LOG_LEVEL: 'error,warn' })); + +const express = require('express'); +const request = require('supertest'); +const { setDataDefinedRoutes, DEFINITION_ROUTES } = require('../router'); + +class Stub { static DataDefined = { definitions: { list: async () => [] }, manifests: {}, runs: () => null }; } + +function mount() { + const a = express(); + a.use(express.json()); + setDataDefinedRoutes(a, { execute: async () => ({ getId: () => 'u1' }) }, { getIntegrationInstance: { execute: async () => null }, integrationClasses: [Stub] }); + return a; +} +// What express actually has mounted, as "METHOD path" (admin middleware is registered with .all, so it is not a method here) +function mounted(a) { + const out = []; + for (const layer of a._router.stack) { + if (!layer.route) continue; + for (const m of Object.keys(layer.route.methods)) if (m !== '_all') out.push(`${m.toUpperCase()} ${layer.route.path}`); + } + return out.sort(); +} + +describe('DEFINITION_ROUTES', () => { + it('lists exactly the routes the router mounts (method and path)', () => { + const listed = Object.values(DEFINITION_ROUTES).map((r) => `${r.method} ${r.path}`).sort(); + expect([...new Set(mounted(mount()))]).toEqual([...new Set(listed)]); + }); + + it('declares method, path and auth for every route, and nothing else', () => { + for (const r of Object.values(DEFINITION_ROUTES)) { + expect(Object.keys(r).sort()).toEqual(['auth', 'method', 'path']); + expect(['user', 'admin']).toContain(r.auth); + } + }); + + it('serves every admin route under /admin/ (never /api/admin) and every user route under /api/', () => { + for (const r of Object.values(DEFINITION_ROUTES)) { + if (r.auth === 'admin') { expect(r.path.startsWith('/admin/integrations/')).toBe(true); expect(r.path).not.toMatch(/^\/api\/admin/); } + else expect(r.path.startsWith('/api/')).toBe(true); + } + expect(Object.values(DEFINITION_ROUTES).filter((r) => r.auth === 'admin').map((r) => `${r.method} ${r.path}`).sort()).toEqual([ + 'GET /admin/integrations/:definitionId/versions', + 'POST /admin/integrations/:definitionId/versions/:version/deprecate', + 'POST /admin/integrations/versions', + ]); + }); + + it('every admin route refuses a request without the admin key, and the old /api/admin paths answer 404', async () => { + process.env.ADMIN_API_KEY = 'k'; + const a = mount(); + expect((await request(a).post('/admin/integrations/versions').send({})).status).toBe(401); + expect((await request(a).get('/admin/integrations/x/versions')).status).toBe(401); + expect((await request(a).post('/admin/integrations/x/versions/1.0.0/deprecate')).status).toBe(401); + expect((await request(a).post('/api/admin/integrations/versions').set('x-frigg-admin-api-key', 'k').send({})).status).toBe(404); + }); +}); diff --git a/packages/core/integrations/data-defined/tests/routes-and-upgrade.test.js b/packages/core/integrations/data-defined/tests/routes-and-upgrade.test.js index 7deead20f..641fd5dbf 100644 --- a/packages/core/integrations/data-defined/tests/routes-and-upgrade.test.js +++ b/packages/core/integrations/data-defined/tests/routes-and-upgrade.test.js @@ -52,7 +52,7 @@ async function setup({ adminKey = 'admin-secret' } = {}) { expect(setDataDefinedRoutes(app, authenticateUser, { getIntegrationInstance, integrationClasses: [Cls] })).toBe(true); app.use((err, req, res, next) => res.status(err.status || 500).json({ error: err.message })); // eslint-disable-line no-unused-vars const install = (id, userId, config, apis = {}) => { const inst = new Cls({ id, config: { type: 'data-defined', ...config } }); Object.assign(inst, apis); installations.set(id, { inst, userId }); return inst; }; - const publishAs = (document, key = adminKey) => request(app).post('/admin/integrations/definitions').set('x-frigg-admin-api-key', key).send({ document }); + const publishAs = (document, key = adminKey) => request(app).post('/admin/integrations/versions').set('x-frigg-admin-api-key', key).send({ document }); return { app, repository, catalog, runStore, install, publishAs, Cls }; } @@ -84,11 +84,11 @@ describe('admin routes: publishing', () => { it('history and deprecation', async () => { const { app, publishAs } = await setup(); await publishAs(example('phone-hubspot')); - const h = await request(app).get('/admin/integrations/definitions/phone-hubspot').set('x-frigg-admin-api-key', 'admin-secret'); + const h = await request(app).get('/admin/integrations/phone-hubspot/versions').set('x-frigg-admin-api-key', 'admin-secret'); expect(h.body.versions.map((x) => x.version)).toEqual(['1.4.0']); - const dep = await request(app).post('/admin/integrations/definitions/phone-hubspot/versions/1.4.0/deprecate').set('x-frigg-admin-api-key', 'admin-secret'); + const dep = await request(app).post('/admin/integrations/phone-hubspot/versions/1.4.0/deprecate').set('x-frigg-admin-api-key', 'admin-secret'); expect(dep.body.version.status).toBe('DEPRECATED'); - expect((await request(app).post('/admin/integrations/definitions/phone-hubspot/versions/1.4.0/deprecate').set('x-frigg-admin-api-key', 'admin-secret')).status).toBe(409); + expect((await request(app).post('/admin/integrations/phone-hubspot/versions/1.4.0/deprecate').set('x-frigg-admin-api-key', 'admin-secret')).status).toBe(409); }); }); diff --git a/packages/core/integrations/data-defined/tests/schedules.test.js b/packages/core/integrations/data-defined/tests/schedules.test.js index c3fc76913..8ed7a9da9 100644 --- a/packages/core/integrations/data-defined/tests/schedules.test.js +++ b/packages/core/integrations/data-defined/tests/schedules.test.js @@ -115,12 +115,12 @@ describe('one shared tick for every installation', () => { /** The tick over the installations' records, as the Lambda runs it (fake repository and queue). */ const tickAt = async (installations, time) => { const sent = []; - const integrationRepository = { findIntegrations: async ({ status }) => installations.filter((i) => i.status === status).map((i) => ({ id: i.id, status: i.status, config: i.config })) }; + const integrationRepository = { findIntegrations: async ({ type, status }) => installations.filter((i) => i.status === status && i.config.type === type).map((i) => ({ id: i.id, status: i.status, config: i.config })) }; const queue = { batchSend: async (entries) => { sent.push(...entries); }, send: async () => {} }; await new RunTriggerTick({ integrationClass: Cls, integrationRepository, queue, queueUrl: QUEUE_URL }).execute({ tick: '5m', time }); return sent; }; - return { Cls, install, tickAt }; + return { Cls, install, tickAt, QUEUE_URL }; }; // The digest runs at 16:00 New York on weekdays: 20:00Z on Monday 2026-10-19 (EDT). const DUE = '2026-10-19T20:00:00Z'; @@ -190,6 +190,30 @@ describe('one shared tick for every installation', () => { expect(await tickAt([inst], DUE)).toEqual([]); }); + it('the shared tick also covers each published definition, an integration type of its own', async () => { + const { Cls, install, tickAt } = setup([digestDef()]); + expect(await Cls.triggerTypes()).toEqual(['data-defined', 'asana-slack']); + const pin = { definition: { id: 'asana-slack', version: digestDef().version } }; + const hosted = install('int7', { type: 'asana-slack', ...pin }, 'ENABLED'); + const legacy = install('int8', pin, 'ENABLED'); // { type: 'data-defined', definition } still works + const due = await tickAt([hosted, legacy], DUE); + expect(due.map((m) => [m.data.integrationId, m.data.trigger]).sort()).toEqual([['int7', DIGEST], ['int8', DIGEST]]); + + // A tick that runs out of time hands on a cursor naming the type it stopped in, and the next one carries on there. + const rows = [legacy, hosted].map((i) => ({ id: i.id, status: i.status, config: i.config })); + const integrationRepository = { findIntegrations: async ({ type, after }) => rows.filter((r) => r.config.type === type && (after == null || r.id > after)) }; + const handedOn = []; + const sent = []; + const queue = { batchSend: async (entries) => { sent.push(...entries); }, send: async (m) => { handedOn.push(m); } }; + const tick = (cursor) => new RunTriggerTick({ integrationClass: Cls, integrationRepository, queue, queueUrl: 'https://sqs/q', pageSize: 1 }).execute({ tick: '5m', time: DUE, cursor, remainingMs: () => 0 }); + await tick(); + expect(handedOn.map((m) => m.data.cursor)).toEqual([{ type: 'data-defined', after: 'int8' }]); + await tick(handedOn[0].data.cursor); + expect(handedOn.map((m) => m.data.cursor)).toEqual([{ type: 'data-defined', after: 'int8' }, { type: 'asana-slack', after: null }]); + await tick(handedOn[1].data.cursor); + expect(sent.map((m) => m.data.integrationId)).toEqual(['int8', 'int7']); + }); + it('a run delivered twice runs once: the trigger and its scheduled time are the dedupe id', async () => { const { install } = setup([digestDef()]); const inst = install('int9', { definition: { id: 'asana-slack', version: digestDef().version } }, 'ENABLED'); diff --git a/packages/core/integrations/data-defined/tests/unification.test.js b/packages/core/integrations/data-defined/tests/unification.test.js new file mode 100644 index 000000000..ee6f8a1e4 --- /dev/null +++ b/packages/core/integrations/data-defined/tests/unification.test.js @@ -0,0 +1,185 @@ +/** + * Published definitions are integration types like any other: each resolves to a class of its own, named by its id, + * listed beside code integrations, installed by { type }, and answered for by the same routes. Only the deployment + * underneath (queue, functions, the app-level webhook URL) is shared. + */ +jest.mock('../../../database/config', () => ({ DB_TYPE: 'mongodb', getDatabaseType: jest.fn(() => 'mongodb'), PRISMA_LOG_LEVEL: 'error,warn' })); + +const express = require('express'); +const request = require('supertest'); +const d = require('..'); +const { setDataDefinedRoutes } = require('../router'); +const { resolveIntegrationClass, listIntegrationClasses } = require('../../integration-types'); +const { GetPossibleIntegrations } = require('../../use-cases/get-possible-integrations'); +const { IntegrationBase } = require('../../integration-base'); + +const manifests = d.loadManifests(); +const example = (id) => JSON.parse(JSON.stringify(d.loadExamples().find((x) => x.id === id))); + +class FakeBase { + constructor(p = {}) { this.events = {}; this.config = p.config; this.id = p.id; this.rows = new Map(); this.queue = []; this.patches = []; } + async initialize() { this.registerEventHandlers(); } + registerEventHandlers() {} + async patchConfig(c) { this.patches.push(c); this.config = { ...this.config, ...c }; } + async onCreate() {} + async getMapping(k) { return this.rows.has(k) ? { mapping: this.rows.get(k) } : null; } + async upsertMapping(k, v) { this.rows.set(k, v); } + async createMapping(k, v) { this.rows.set(k, v); } + async deleteMapping(k) { this.rows.delete(k); } + async queueWebhook(m) { this.queue.push(m); } + static getDeploymentName() { return (this.Host || this).Definition.name; } +} + +/** A code integration, as an app registers one. */ +class HubspotSlack extends FakeBase { + static Definition = { name: 'hubspot-slack-code', version: '3.0.0', modules: { hubspot: { definition: { getName: () => 'hubspot' } } }, display: { label: 'Code one' } }; + static getOptionDetails() { return { type: 'hubspot-slack-code', display: { name: 'Code one' } }; } +} + +function host(docs = [example('phone-hubspot')], Base = FakeBase) { + return d.defineSharedIntegrationClass({ definitions: d.memoryDefinitions(docs), manifests, modules: { phone: { getName: () => 'phone' }, hubspot: { getName: () => 'hubspot' } }, IntegrationBase: Base, runs: false }); +} + +describe('integration types', () => { + it('a published definition resolves to its own class: its name, its label, the host underneath', async () => { + const Host = host(); + const Cls = await resolveIntegrationClass([HubspotSlack, Host], 'phone-hubspot'); + expect(Cls.Definition).toMatchObject({ name: 'phone-hubspot', version: '1.4.0', display: { label: 'Phone ↔ HubSpot', category: 'CRM' } }); + expect(Object.keys(Cls.Definition.modules)).toEqual(['phone', 'hubspot']); + expect(Cls.Host).toBe(Host); + expect(Cls.getDeploymentName()).toBe('data-defined'); + expect(Cls.IntegrationProvider).toBeUndefined(); + expect(await resolveIntegrationClass([HubspotSlack, Host], 'phone-hubspot')).toBe(Cls); // one class per type + expect(await resolveIntegrationClass([HubspotSlack, Host], 'hubspot-slack-code')).toBe(HubspotSlack); + expect(await resolveIntegrationClass([HubspotSlack, Host], 'nothing-by-this-name')).toBeNull(); + expect(await resolveIntegrationClass([HubspotSlack, Host], 'data-defined')).toBe(Host); // legacy installs keep working + }); + + it('code wins: a definition with a registered name never shadows the class', async () => { + const doc = { ...example('phone-hubspot'), id: 'hubspot-slack-code' }; + expect(await resolveIntegrationClass([HubspotSlack, host([doc])], 'hubspot-slack-code')).toBe(HubspotSlack); + expect((await listIntegrationClasses([HubspotSlack, host([doc])])).map((c) => c.Definition.name)).toEqual(['hubspot-slack-code']); + }); + + it('a definition needing an app module this deployment lacks is not installable here', async () => { + const Host = host([example('phone-hubspot'), example('shopify-netsuite')]); + expect(await resolveIntegrationClass([Host], 'shopify-netsuite')).toBeNull(); + expect((await listIntegrationClasses([Host])).map((c) => c.Definition.name)).toEqual(['phone-hubspot']); + }); + + it('GET /api/integrations options list code and published types the same way; the host itself is not one', async () => { + const options = await new GetPossibleIntegrations({ integrationClasses: [HubspotSlack, host()] }).execute(); + expect(options).toEqual([ + { type: 'hubspot-slack-code', display: { name: 'Code one' } }, + { type: 'phone-hubspot', hasUserConfig: true, isMany: false, requiresNewEntity: false, display: { name: 'Phone ↔ HubSpot', description: expect.any(String), detailsUrl: undefined, icon: undefined } }, + ]); + }); +}); + +describe('installing by type', () => { + it('{ type } pins the latest version at install, and keeps it when a newer one is published', async () => { + const v1 = example('phone-hubspot'); + const docs = [v1]; + const Host = host(docs); + const Cls = await resolveIntegrationClass([Host], 'phone-hubspot'); + const inst = new Cls({ id: 'int9', config: { type: 'phone-hubspot', settings: { fromNumber: '+15125550100' } } }); + inst.phone = { api: { createWebhook: async () => ({ id: 'W1' }) } }; + inst.hubspot = { api: {}, entity: { externalId: '1' } }; + await inst.initialize(); + await inst.onCreate({}); + expect(inst.patches[0].definition).toEqual({ id: 'phone-hubspot', version: '1.4.0', checksum: expect.any(String) }); + + docs.push({ ...JSON.parse(JSON.stringify(v1)), version: '1.5.0' }); + const later = new Cls({ id: 'int9', config: inst.config }); + await later.initialize(); + expect(later.def.version).toBe('1.4.0'); + }); + + it('refuses an installation whose pin names another definition than its type', async () => { + const Cls = await resolveIntegrationClass([host([example('phone-hubspot'), example('github-linear')])], 'phone-hubspot'); + await expect(new Cls({ id: 'x', config: { type: 'phone-hubspot', definition: { id: 'github-linear', version: '1.0.0' } } }).initialize()).rejects.toThrow('type is "phone-hubspot"'); + }); + + it('webhooks: its own URL for its own webhooks; the shared URL for app-level ones', async () => { + const Cls = await resolveIntegrationClass([host()], 'phone-hubspot'); + const inst = new Cls({ id: 'int9', config: { type: 'phone-hubspot', definition: { id: 'phone-hubspot', version: '1.4.0' } } }); + await inst.initialize(); + const { webhookUrl, webhookUrls } = inst.installation(); + expect(webhookUrl).toBe('/api/phone-hubspot-integration/webhooks/int9'); + expect(webhookUrls.phone).toBe('/api/phone-hubspot-integration/webhooks/int9?module=phone'); + expect(webhookUrls.hubspot).toBe('/api/data-defined-integration/webhooks?module=hubspot'); // HubSpot posts every account to one URL + }); +}); + +describe('on the shared deployment', () => { + const OLD = { ...process.env }; + afterEach(() => { process.env = { ...OLD }; }); + + it('queues and schedules on the host queue, while reporting its own name', async () => { + process.env.DATA_DEFINED_QUEUE_URL = 'https://sqs/host-queue'; + const Host = host([example('phone-hubspot')], IntegrationBase); + const Cls = await resolveIntegrationClass([Host], 'phone-hubspot'); + expect(Cls.getDeploymentName()).toBe('data-defined'); + expect(IntegrationBase.getDeploymentName.call(HubspotSlack)).toBe('hubspot-slack-code'); + const inst = new Cls({ id: 'int9', config: { type: 'phone-hubspot', definition: { id: 'phone-hubspot', version: '1.4.0' } } }); + expect(inst.scheduleQueueUrl()).toBe('https://sqs/host-queue'); + expect(inst.getTelemetryContext().integrationType).toBe('phone-hubspot'); + const { QueuerUtil } = require('../../../queues'); + const send = jest.spyOn(QueuerUtil, 'send').mockResolvedValue('m1'); + await inst.queueWebhook({ event: 'ON_WEBHOOK', integrationId: 'int9' }); + expect(send).toHaveBeenCalledWith(expect.objectContaining({ event: 'ON_WEBHOOK' }), 'https://sqs/host-queue'); + send.mockRestore(); + }); + + it('app-level webhooks fan out to installations of every type the host provides, and no others', async () => { + const lookups = []; + const Host = d.defineSharedIntegrationClass({ + definitions: d.memoryDefinitions([example('phone-hubspot')]), manifests, modules: { phone: {}, hubspot: {} }, IntegrationBase: FakeBase, runs: false, + lookup: async (account, moduleName) => { lookups.push([account, moduleName]); return ['a1']; }, + }); + const dry = new Host(); + await dry.initialize(); + const out = await dry.onWebhook({ data: { body: [{ portalId: 12345678, objectId: 5, subscriptionType: 'contact.creation', eventId: 1 }], headers: {}, query: { module: 'hubspot' } } }); + expect(out).toEqual({ queued: 1 }); + expect(lookups).toEqual([['12345678', 'hubspot']]); + // A dry instance of a published type (no installation yet) fans out too, rather than routing as if installed. + const Cls = await resolveIntegrationClass([Host], 'phone-hubspot'); + const dryTyped = new Cls(); + await dryTyped.initialize(); + expect(await dryTyped.onWebhook({ data: { body: [{ portalId: 12345678, objectId: 5, subscriptionType: 'contact.creation', eventId: 1 }], headers: {}, query: { module: 'hubspot' } } })).toEqual({ queued: 1 }); + }); +}); + +describe('the same routes for every integration', () => { + function app(inst) { + const a = express(); + a.use(express.json()); + const authenticateUser = { execute: async () => ({ getId: () => 'u1' }) }; + const getIntegrationInstance = { integrationRepository: { findIntegrationById: async (id) => ({ id, userId: 'u1' }) }, execute: async () => inst }; + setDataDefinedRoutes(a, authenticateUser, { getIntegrationInstance, integrationClasses: [HubspotSlack, host()] }); + a.use((err, req, res, next) => res.status(err.status || 500).json({ error: err.message })); // eslint-disable-line no-unused-vars + return a; + } + + it('a code integration answers runs, syncs, upgrade and decision with what it has, never "not data-defined"', async () => { + const code = new HubspotSlack({ id: 'c1', config: { type: 'hubspot-slack-code' } }); + expect((await request(app(code)).get('/api/integrations/c1/runs')).body).toEqual({ runs: [] }); + expect((await request(app(code)).get('/api/integrations/c1/syncs')).body).toEqual({ syncs: [] }); + const plan = await request(app(code)).get('/api/integrations/c1/upgrade'); + expect(plan.status).toBe(200); + expect(plan.body).toMatchObject({ allowed: false, from: '3.0.0', to: '3.0.0' }); + expect((await request(app(code)).post('/api/integrations/c1/upgrade').send({})).status).toBe(409); + const decision = await request(app(code)).post('/api/integrations/c1/runs/r1/decision').send({ decision: 'approve' }); + expect(decision.status).toBe(404); + expect(decision.body.code).toBe('NOT_WAITING'); + }); + + it('publishing answers at /admin/integrations/versions, and nowhere else', async () => { + process.env.ADMIN_API_KEY = 'admin-key'; + const a = app(new HubspotSlack({ id: 'c1' })); + const r = await request(a).post('/admin/integrations/versions').set('x-frigg-admin-api-key', 'admin-key').send({ document: example('phone-hubspot') }); + expect(r.status).toBe(501); // bundled definitions: the route exists, there is no store + expect(r.body.code).toBe('NO_STORE'); + expect((await request(a).post('/api/admin/data-defined/definitions').set('x-frigg-admin-api-key', 'admin-key').send({})).status).toBe(404); + }); +}); diff --git a/packages/core/integrations/data-defined/tests/workspace.test.js b/packages/core/integrations/data-defined/tests/workspace.test.js new file mode 100644 index 000000000..04fbaff7c --- /dev/null +++ b/packages/core/integrations/data-defined/tests/workspace.test.js @@ -0,0 +1,117 @@ +/** + * Integrations a workspace builds: published by a signed-in user into their own + * workspace, listed and installable there like any other, invisible to everyone else, and read back for the editor. + */ +const express = require('express'); +const request = require('supertest'); +const d = require('..'); +const { setDataDefinedRoutes } = require('../router'); +const { resolveIntegrationClass, listIntegrationClasses } = require('../../integration-types'); +const { CreateIntegration } = require('../../use-cases/create-integration'); + +const manifests = d.loadManifests(); +const example = (id) => JSON.parse(JSON.stringify(d.loadExamples().find((x) => x.id === id))); + +/** The repository contract (definition-version-repository.js), in memory. */ +function memoryRepository() { + const rows = []; + let n = 0; + const copy = (r) => (r ? JSON.parse(JSON.stringify(r)) : null); + return { + rows, + async findVersion(id, version) { return copy(rows.find((r) => r.definitionId === id && r.version === version)); }, + async listVersions(id) { return rows.filter((r) => r.definitionId === id).map(copy); }, + async listByStatus(statuses) { return rows.filter((r) => statuses.includes(r.status)).map(({ document, ...r }) => copy(r)); }, + async createVersion(data) { const row = { ...copy(data), id: String(++n) }; rows.push(row); return copy(row); }, + async updateIfStatus(id, version, from, data) { const r = rows.find((x) => x.definitionId === id && x.version === version && x.status === from); if (!r) return null; Object.assign(r, copy(data)); return copy(r); }, + }; +} + +class FakeBase { + constructor(p = {}) { this.events = {}; this.config = p.config; this.id = p.id; } + async initialize() {} + registerEventHandlers() {} + static getDeploymentName() { return (this.Host || this).Definition.name; } +} +class CodeOne extends FakeBase { static Definition = { name: 'code-one', modules: {} }; static getOptionDetails() { return { type: 'code-one' }; } } + +function setup() { + const repository = memoryRepository(); + const catalog = d.createDefinitionCatalog({ repository }); + const Host = d.defineSharedIntegrationClass({ definitions: catalog, manifests, modules: { phone: {}, hubspot: {}, github: {}, linear: {}, slack: {} }, IntegrationBase: FakeBase, runs: false }); + const classes = [CodeOne, Host]; + const app = (userId) => { + const a = express(); + a.use(express.json()); + setDataDefinedRoutes(a, { execute: async () => ({ getId: () => userId }) }, { getIntegrationInstance: { execute: async () => null }, integrationClasses: classes }); + a.use((err, req, res, next) => res.status(err.status || 500).json({ error: err.message })); // eslint-disable-line no-unused-vars + return a; + }; + return { repository, catalog, classes, app }; +} + +const own = (over = {}) => ({ ...example('github-linear'), id: 'acme-issues', version: '1.0.0', ...over }); + +describe("a workspace's own integrations", () => { + it('a signed-in user publishes one into their workspace; it then lists and installs there like any other', async () => { + const { app, classes, repository } = setup(); + await new d.PublishDefinitionVersion({ repository, manifests }).execute({ document: example('phone-hubspot') }); // the app's catalogue + const r = await request(app('u-acme')).post('/api/integration-definitions').send({ document: own() }); + expect(r.status).toBe(201); + expect(r.body.version).toMatchObject({ definitionId: 'acme-issues', workspaceId: 'u-acme', createdBy: 'u-acme', status: 'PUBLISHED' }); + expect(r.body.version.document).toBeUndefined(); + + const mine = (await listIntegrationClasses(classes, { workspaceId: 'u-acme' })).map((c) => c.getOptionDetails()); + expect(mine.map((o) => [o.type, Boolean(o.editable)])).toEqual([['code-one', false], ['acme-issues', true], ['phone-hubspot', false]]); + expect((await resolveIntegrationClass(classes, 'acme-issues', { workspaceId: 'u-acme' })).Definition.name).toBe('acme-issues'); + }); + + it('no one else can see, install or read it; an existing installation still hydrates', async () => { + const { app, classes } = setup(); + await request(app('u-acme')).post('/api/integration-definitions').send({ document: own() }); + expect((await listIntegrationClasses(classes, { workspaceId: 'u-other' })).map((c) => c.Definition.name)).toEqual(['code-one']); + expect(await resolveIntegrationClass(classes, 'acme-issues', { workspaceId: 'u-other' })).toBeNull(); + expect((await resolveIntegrationClass(classes, 'acme-issues')).Definition.name).toBe('acme-issues'); // the worker hydrating by id + expect((await request(app('u-other')).get('/api/data-defined/definitions')).body.definitions).toEqual([]); + expect((await request(app('u-acme')).get('/api/data-defined/definitions')).body.definitions.map((x) => x.id)).toEqual(['acme-issues']); + const read = await request(app('u-other')).get('/api/integration-definitions/acme-issues'); + expect(read.status).toBe(404); + + const create = new CreateIntegration({ integrationRepository: { findIntegrationsByUserId: async () => [], createIntegration: async () => { throw new Error('must not create'); } }, integrationClasses: classes, moduleFactory: {} }); + await expect(create.execute(['e1'], 'u-other', { type: 'acme-issues' })).rejects.toThrow('No integration class found for type: acme-issues'); + }); + + it("the provider itself hides another workspace's integration, even from a store that lists everything", async () => { + const doc = own(); + const store = { ...d.memoryDefinitions([doc]), list: async () => [{ id: doc.id, version: doc.version, display: doc.display, modules: { github: 'github', linear: 'linear', slack: 'slack' }, workspaceId: 'u-acme' }], ownerOf: async () => 'u-acme' }; + const Host = d.defineSharedIntegrationClass({ definitions: store, manifests, modules: { github: {}, linear: {}, slack: {} }, IntegrationBase: FakeBase, runs: false }); + expect((await listIntegrationClasses([Host], { workspaceId: 'u-other' })).map((c) => c.Definition.name)).toEqual([]); + expect((await listIntegrationClasses([Host], { workspaceId: 'u-acme' })).map((c) => c.Definition.name)).toEqual(['acme-issues']); + }); + + it('the editor reads every version, drafts included, and the newest document', async () => { + const { app } = setup(); + await request(app('u-acme')).post('/api/integration-definitions').send({ document: own() }); + await request(app('u-acme')).post('/api/integration-definitions').send({ document: own({ version: '1.1.0', display: { ...own().display, label: 'Acme issues v2' } }), publish: false }); + const r = await request(app('u-acme')).get('/api/integration-definitions/acme-issues'); + expect(r.status).toBe(200); + expect(r.body.versions.map((v) => [v.version, v.status])).toEqual([['1.1.0', 'DRAFT'], ['1.0.0', 'PUBLISHED']]); + expect(r.body.document.display.label).toBe('Acme issues v2'); + }); + + it("an id has one owner: not another workspace's, not the app's catalogue, not a code integration's", async () => { + const { app, repository } = setup(); + await new d.PublishDefinitionVersion({ repository, manifests }).execute({ document: example('phone-hubspot') }); + await request(app('u-acme')).post('/api/integration-definitions').send({ document: own() }); + const taken = async (userId, document) => { const r = await request(app(userId)).post('/api/integration-definitions').send({ document }); return [r.status, r.body.code]; }; + // Another workspace's id answers as a miss (it must not confirm the id exists), the same as reading it does. + expect(await taken('u-other', own({ version: '2.0.0' }))).toEqual([404, 'NOT_FOUND']); + const foreignPublish = await request(app('u-other')).post('/api/integration-definitions').send({ document: own({ version: '2.0.0' }) }); + const foreignRead = await request(app('u-other')).get('/api/integration-definitions/acme-issues'); + expect([foreignPublish.status, foreignPublish.body]).toEqual([foreignRead.status, foreignRead.body]); + expect(JSON.stringify(foreignPublish.body)).not.toMatch(/already|taken|another/i); + expect(await taken('u-acme', { ...example('phone-hubspot'), version: '9.0.0' })).toEqual([409, 'ID_TAKEN']); + expect(await taken('u-acme', own({ id: 'code-one' }))).toEqual([409, 'ID_TAKEN']); + expect(await taken('u-acme', own({ version: '1.0.1' }))).toEqual([201, undefined]); // its own next version is fine + }); +}); diff --git a/packages/core/integrations/data-defined/use-cases/definition-use-cases.js b/packages/core/integrations/data-defined/use-cases/definition-use-cases.js index beb35d3d1..68cbcd54b 100644 --- a/packages/core/integrations/data-defined/use-cases/definition-use-cases.js +++ b/packages/core/integrations/data-defined/use-cases/definition-use-cases.js @@ -55,30 +55,54 @@ function createDefinitionCatalog({ repository } = {}) { async checksumOf(definitionId, version) { return (await repo().findVersion(definitionId, version))?.checksum || null; }, - /** The install catalogue: latest published version of each definition. */ - list: () => new ListPublishedDefinitions({ repository: repo() }).execute(), + /** The workspace that built a definition, or null for the app's own catalogue. */ + async ownerOf(definitionId) { + const rows = await repo().listVersions(definitionId); + return rows[0]?.workspaceId ?? null; + }, + /** The install catalogue: latest published version of each definition (with `workspaceId`, that workspace's too). */ + list: (context) => new ListPublishedDefinitions({ repository: repo() }).execute(context), /** The repository behind this catalogue, for publishing (bundled stores have none). */ get repository() { return repo(); }, }; } +/** The one answer for an integration id that is not this workspace's, whether it is someone else's or does not exist. */ +const notInWorkspace = (id) => new DefinitionError(`No integration "${id}" in this workspace`, { status: 404, code: 'NOT_FOUND' }); + class PublishDefinitionVersion { /** * @param {{ repository, manifests: object, check?: (def, manifests) => { valid, errors, warnings } }} deps * check defaults to schema-check's checkDefinition (JSON Schema + references) */ - constructor({ repository, manifests, check }) { + constructor({ repository, manifests, check, isReserved }) { this.repository = repository; this.manifests = manifests; this.check = check || require('../schema-check').checkDefinition; + /** Names a definition may not take: the app's registered (code) integrations. */ + this.isReserved = isReserved || (() => false); } /** @returns {Promise<{ version: object, created: boolean, changeLevel: string|null, warnings: array }>} */ - async execute({ document, notes, createdBy, publish = true, strict = false } = {}) { + /** @param {{ workspaceId?: string }} p workspaceId: the workspace building its own integration (null: the app's catalogue) */ + async execute({ document, notes, createdBy, publish = true, strict = false, workspaceId = null } = {}) { if (!document || typeof document !== 'object') throw new DefinitionError('Send the definition as `document`'); const result = this.check(document, this.manifests); if (!result.valid || (strict && result.warnings.length)) throw new DefinitionError(`${document.id}@${document.version} is not valid`, { code: 'INVALID', details: { errors: result.errors, warnings: result.warnings } }); + // An id belongs to one owner: the app's catalogue, or one workspace. Nobody publishes over someone else's, nor + // takes the name of a code integration. + if (this.isReserved(document.id)) throw new DefinitionError(`"${document.id}" is the name of an integration this app already has`, { status: 409, code: 'ID_TAKEN' }); + const versions = await this.repository.listVersions(document.id); + const other = versions.find((v) => (v.workspaceId ?? null) !== (workspaceId ?? null)); + if (other) { + // Another WORKSPACE's id must not be confirmed to this one: answer as a miss, exactly as reading it does + // (GetOwnDefinition; ADR-052 section 3: a foreign resource is a 404, so ids cannot be probed). The catalogue's + // ids are public (they are listed), and the admin is not a tenant, so those may say what is wrong. + if (workspaceId != null && other.workspaceId != null) throw notInWorkspace(document.id); + throw new DefinitionError(`"${document.id}" is already used by another integration`, { status: 409, code: 'ID_TAKEN' }); + } + const sum = checksum(document); const existing = await this.repository.findVersion(document.id, document.version); if (existing && existing.status !== 'DRAFT') { @@ -86,7 +110,7 @@ class PublishDefinitionVersion { throw new DefinitionError(`${document.id}@${document.version} is already ${existing.status.toLowerCase()} and cannot change. Publish a new version.`, { status: 409, code: 'IMMUTABLE' }); } - const published = newestFirst((await this.repository.listVersions(document.id)).filter((r) => r.status === 'PUBLISHED' || r.status === 'DEPRECATED')); + const published = newestFirst(versions.filter((r) => r.status === 'PUBLISHED' || r.status === 'DEPRECATED')); const previous = published.find((r) => r.version !== document.version) || null; let changeLevel = null; if (previous) { @@ -98,7 +122,7 @@ class PublishDefinitionVersion { const row = { definitionId: document.id, version: document.version, schemaVersion: document.schemaVersion, checksum: sum, document, - changeLevel, previousVersion: previous?.version || null, notes: notes ?? null, createdBy: createdBy ?? null, + changeLevel, previousVersion: previous?.version || null, notes: notes ?? null, createdBy: createdBy ?? null, workspaceId: workspaceId ?? null, status: publish ? 'PUBLISHED' : 'DRAFT', publishedAt: publish ? new Date() : null, }; if (existing) { @@ -135,10 +159,14 @@ class DeprecateDefinitionVersion { class ListPublishedDefinitions { constructor({ repository }) { this.repository = repository; } - /** Latest published version of each definition, with what an install screen shows. */ - async execute() { + /** + * Latest published version of each definition, with what an install screen shows: the app's catalogue, plus the + * integrations `workspaceId` built (another workspace's are never listed). Without a workspace, every one. + */ + async execute({ workspaceId } = {}) { const byId = new Map(); for (const r of await this.repository.listByStatus(['PUBLISHED'])) { + if (workspaceId && r.workspaceId && r.workspaceId !== String(workspaceId)) continue; const cur = byId.get(r.definitionId); if (!cur || compareVersions(r.version, cur.version) > 0) byId.set(r.definitionId, r); } @@ -146,7 +174,7 @@ class ListPublishedDefinitions { for (const r of byId.values()) { const doc = (await this.repository.findVersion(r.definitionId, r.version)).document; out.push({ - id: r.definitionId, version: r.version, checksum: r.checksum, publishedAt: r.publishedAt, + id: r.definitionId, version: r.version, checksum: r.checksum, publishedAt: r.publishedAt, workspaceId: r.workspaceId ?? null, display: doc.display, modules: Object.fromEntries(Object.entries(doc.modules || {}).map(([k, m]) => [k, m.module])), counts: { syncs: (doc.syncs || []).length, links: (doc.links || []).length, workflows: (doc.workflows || []).length, ui: (doc.ui || []).length }, }); @@ -164,6 +192,35 @@ class GetDefinitionHistory { } } +/** + * The installation an approval decision is about, for someone who may not own it: an approval step names its + * approvers, and they decide from their own sign-in. Whether this user may decide a particular run is the run's + * business (decide()); here, only who owns it. + */ +/** + * A workspace's own integration, for its editor: every version (drafts included) and the newest document. Anyone + * else's, or the app's catalogue, is "not found": the editor exists only for what this workspace built. + */ +class GetOwnDefinition { + constructor({ repository }) { this.repository = repository; } + async execute({ definitionId, workspaceId }) { + const rows = newestFirst(await this.repository.listVersions(definitionId)); + if (!rows.length || !workspaceId || rows[0].workspaceId !== String(workspaceId)) throw notInWorkspace(definitionId); + return { id: definitionId, versions: rows.map(({ document, ...meta }) => meta), document: rows[0].document }; + } +} + +class GetIntegrationForDecision { + constructor({ getIntegrationInstance }) { this.getIntegrationInstance = getIntegrationInstance; } + async execute(integrationId, userId) { + const record = await this.getIntegrationInstance.integrationRepository.findIntegrationById(integrationId).catch(() => null); + if (!record) throw new DefinitionError(`Integration ${integrationId} not found`, { status: 404, code: 'NOT_FOUND' }); + const isOwner = String(record.userId) === String(userId); + const integration = await this.getIntegrationInstance.execute(integrationId, record.userId); + return { integration, isOwner }; + } +} + class PlanInstallationUpgrade { constructor({ catalog }) { this.catalog = catalog; } /** @returns {Promise<{ from, to, allowed, needs, actions, config, changes }>} */ @@ -184,4 +241,5 @@ class PlanInstallationUpgrade { module.exports = { DefinitionError, compareVersions, createDefinitionCatalog, PublishDefinitionVersion, DeprecateDefinitionVersion, ListPublishedDefinitions, GetDefinitionHistory, PlanInstallationUpgrade, + GetIntegrationForDecision, GetOwnDefinition, }; diff --git a/packages/core/integrations/data-defined/validate.js b/packages/core/integrations/data-defined/validate.js index d8908cb85..b0429d9dd 100644 --- a/packages/core/integrations/data-defined/validate.js +++ b/packages/core/integrations/data-defined/validate.js @@ -23,7 +23,12 @@ const TRIGGERS = new Set(['webhook', 'schedule', 'userAction', 'syncRecord']); const TRANSFORMS = new Set(Object.keys(TRANSFORM_FNS)); const INSTALL_KEYS = new Set(['webhookUrl', 'webhookUrls', 'portalId', 'runtime', 'integrationId', 'accounts']); -function validate(def, manifests = {}) { +/** + * @param {object} [options] + * @param {string[]} [options.codeMethods] the methods of the code class a definition is embedded in + * (extendIntegrationClass): a code step may call them as `#method`. A published definition has none. + */ +function validate(def, manifests = {}, { codeMethods } = {}) { const errors = [], warnings = []; const err = (path, message) => errors.push({ path, message }); const warn = (path, message) => warnings.push({ path, message }); @@ -288,7 +293,10 @@ function validate(def, manifests = {}) { if (st.use === 'map' && Array.isArray(args.mapping)) checkRows(args.mapping, `${p}.with.mapping`); if (st.use === 'template' && args.id && !def.templates?.[args.id]) err(`${p}.with.id`, `Unknown template "${args.id}"`); if (st.use === 'map' && args.sync && !syncs[args.sync]) err(`${p}.with.sync`, `Unknown sync "${args.sync}"`); - if (st.use === 'code' && !/^[@\w/.-]+#\w+@\d+\.\d+\.\d+$/.test(args.ref || '')) err(`${p}.with.ref`, 'Code steps are "package#function@x.y.z"'); + if (st.use === 'code' && /^#/.test(args.ref || '')) { + if (!codeMethods) err(`${p}.with.ref`, `"${args.ref}" calls a method of a code integration; only a definition embedded in one can (extendIntegrationClass)`); + else if (!codeMethods.includes(String(args.ref).slice(1))) err(`${p}.with.ref`, `The integration has no method "${String(args.ref).slice(1)}"`); + } else if (st.use === 'code' && !/^[@\w/.-]+#\w+@\d+\.\d+\.\d+$/.test(args.ref || '')) err(`${p}.with.ref`, 'Code steps are "package#function@x.y.z" (or "#method" in a definition a code integration embeds)'); } else { const cap = splitCapability(st.use); const man = cap && manifestOf[cap.module]; diff --git a/packages/core/integrations/integration-base.js b/packages/core/integrations/integration-base.js index 24965d364..860990cfa 100644 --- a/packages/core/integrations/integration-base.js +++ b/packages/core/integrations/integration-base.js @@ -109,6 +109,14 @@ class IntegrationBase { return this.Definition.name; } + /** + * The deployed integration whose queue and functions this one runs on: itself, unless another class hosts it + * (`static Host`, e.g. a published integration definition run by the shared runtime). + */ + static getDeploymentName() { + return (this.Host || this).Definition.name; + } + static getCurrentVersion() { return this.Definition.version; } @@ -585,6 +593,12 @@ class IntegrationBase { return outcome; } + /** The queue a one-time scheduled job (a paused run's resume) fires into: this integration's deployment queue (`_QUEUE_URL`). */ + scheduleQueueUrl() { + if (typeof this.getQueueUrl === 'function') return this.getQueueUrl(); + return process.env[`${this.constructor.getDeploymentName().toUpperCase().replace(/-/g, '_')}_QUEUE_URL`]; + } + /** A tick that ran out of time continues here, on a dry instance, from its page cursor. */ async onTriggerTick({ data = {}, context } = {}) { const { runTriggerTick } = require('./use-cases/run-trigger-tick'); @@ -710,7 +724,7 @@ class IntegrationBase { async queueWebhook(data) { const { QueuerUtil } = require('../queues'); - const queueName = `${this.constructor.Definition.name + const queueName = `${this.constructor.getDeploymentName() .toUpperCase() .replace(/-/g, '_')}_QUEUE_URL`; const queueUrl = process.env[queueName]; diff --git a/packages/core/integrations/integration-router.js b/packages/core/integrations/integration-router.js index 11576672b..bf46ff700 100644 --- a/packages/core/integrations/integration-router.js +++ b/packages/core/integrations/integration-router.js @@ -209,7 +209,7 @@ function createIntegrationRouter() { updateIntegration, getPossibleIntegrations, }); - // Data-defined integrations (ADR-063): only mounted when the app registers a data-defined class. + // Integration versions, run history, sync progress and approvals (ADR-063): mounted when the app registers the integration runtime. require('./data-defined/router').setDataDefinedRoutes(router, authenticateUser, { getIntegrationInstance, integrationClasses, @@ -274,7 +274,7 @@ function setIntegrationRoutes(router, authenticateUser, useCases) { const integrations = await getIntegrationsForUser.execute(userId); const results = { entities: { - options: await getPossibleIntegrations.execute(), + options: await getPossibleIntegrations.execute({ userId }), authorized: await getEntitiesForUser.execute(userId), }, integrations: integrations, diff --git a/packages/core/integrations/integration-types.js b/packages/core/integrations/integration-types.js new file mode 100644 index 000000000..688a66b3c --- /dev/null +++ b/packages/core/integrations/integration-types.js @@ -0,0 +1,64 @@ +/** + * Integration types: how an installation's `config.type` becomes the class that runs it. + * + * Most types are classes the app registers at startup (appDefinition.integrations). A registered class may also be + * a provider: `static IntegrationProvider = { resolve(type), list() }` answers for types that exist as data rather + * than code (published integration definitions). Those resolve to classes of their own, with their own name, so a + * caller cannot tell how an integration was made; only the deployment underneath (queue, functions) is shared. + * + * Registered classes win over provided ones, so a published definition can never shadow code. When one would have, + * that is logged (once per name per process) as `frigg.integrations.definition.shadowed_by_code`: the definition is + * dead weight and whoever published it expects it to be installable. + */ +const { getLogger } = require('../logs'); + +const warnedShadowed = new Set(); + +/** A definition is ignored because registered code has the same name. Logging must never break a listing. */ +function warnShadowed(name, provider) { + if (warnedShadowed.has(name)) return; + warnedShadowed.add(name); + try { + getLogger('frigg.integrations').warn('A published integration definition is ignored because registered code has the same name', { + eventName: 'frigg.integrations.definition.shadowed_by_code', + integrationName: name, + provider: provider?.name || null, + }); + } catch { /* ignore */ } +} + +/** + * The class for a type, or null. + * @param {object} [context] who is asking. With `workspaceId`, a type built in another workspace does not resolve: + * what a user may install or see. Without it (hydrating an installation that exists) every type resolves. + */ +async function resolveIntegrationClass(integrationClasses, type, context) { + if (!type) return null; + const registered = integrationClasses.find((c) => c?.Definition?.name === type); + if (registered) return registered; + for (const c of integrationClasses) { + if (!c?.IntegrationProvider) continue; + const provided = await c.IntegrationProvider.resolve(type, context); + if (provided) return provided; + } + return null; +} + +/** + * Every type a user can install: the registered classes, except providers (which are hosts, not integrations), + * plus every type each provider offers. + */ +async function listIntegrationClasses(integrationClasses, context) { + const registered = new Set(integrationClasses.map((c) => c?.Definition?.name)); + const out = []; + for (const c of integrationClasses) { + if (!c?.IntegrationProvider) { out.push(c); continue; } + for (const provided of await c.IntegrationProvider.list(context)) { + if (registered.has(provided.Definition.name)) warnShadowed(provided.Definition.name, c); + else out.push(provided); + } + } + return out; +} + +module.exports = { resolveIntegrationClass, listIntegrationClasses, _resetShadowWarningsForTests: () => warnedShadowed.clear() }; diff --git a/packages/core/integrations/tests/integration-types-shadow.test.js b/packages/core/integrations/tests/integration-types-shadow.test.js new file mode 100644 index 000000000..8bb0278d6 --- /dev/null +++ b/packages/core/integrations/tests/integration-types-shadow.test.js @@ -0,0 +1,35 @@ +const { listIntegrationClasses, resolveIntegrationClass, _resetShadowWarningsForTests } = require('../integration-types'); +const { createMemorySink } = require('../../logs'); + +const code = (name) => class { static Definition = { name }; }; +const host = (names) => class Host { + static Definition = { name: 'host' }; + static IntegrationProvider = { resolve: async () => null, list: async () => names.map((n) => code(n)) }; +}; + +describe('a published definition shadowed by registered code', () => { + beforeEach(() => _resetShadowWarningsForTests()); + + it('is ignored (code wins) and logged once as frigg.integrations.definition.shadowed_by_code', async () => { + const sink = createMemorySink(); + const classes = [code('clash'), host(['clash', 'fine'])]; + const first = await listIntegrationClasses(classes); + await listIntegrationClasses(classes); + expect(first.map((c) => c.Definition.name).sort()).toEqual(['clash', 'fine']); + expect(first.find((c) => c.Definition.name === 'clash')).toBe(classes[0]); + const warns = sink.records.filter((r) => r.eventName === 'frigg.integrations.definition.shadowed_by_code'); + expect(warns).toHaveLength(1); + expect(warns[0]).toMatchObject({ level: 'WARN', integrationName: 'clash', logger: 'frigg.integrations' }); + }); + + it('says nothing when no name clashes', async () => { + const sink = createMemorySink(); + await listIntegrationClasses([code('a'), host(['b'])]); + expect(sink.records.filter((r) => String(r.eventName).includes('shadowed'))).toHaveLength(0); + }); + + it('still resolves to the code class', async () => { + const classes = [code('clash'), host(['clash'])]; + expect(await resolveIntegrationClass(classes, 'clash')).toBe(classes[0]); + }); +}); diff --git a/packages/core/integrations/use-cases/create-integration.js b/packages/core/integrations/use-cases/create-integration.js index f0c06c106..c8c2b8c5f 100644 --- a/packages/core/integrations/use-cases/create-integration.js +++ b/packages/core/integrations/use-cases/create-integration.js @@ -1,3 +1,4 @@ +const { resolveIntegrationClass } = require('../integration-types'); // Removed Integration wrapper - using IntegrationBase directly const { mapIntegrationClassToIntegrationDTO, @@ -38,6 +39,16 @@ class CreateIntegration { `[Frigg] Creating ${config?.type} integration for user ${userId} with entities [${entities}]` ); + // Only a type this user may install: registered code, the app's catalogue, or one their workspace built. + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + config?.type, + { workspaceId: String(userId) } + ); + if (!integrationClass) { + throw new Error(`No integration class found for type: ${config?.type}`); + } + const existing = await this._findDuplicate( userId, config?.type, @@ -174,10 +185,9 @@ class CreateIntegration { } async _buildInstance(integrationRecord) { - const integrationClass = this.integrationClasses.find( - (integrationClass) => - integrationClass.Definition.name === - integrationRecord.config.type + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + integrationRecord.config.type ); if (!integrationClass) { diff --git a/packages/core/integrations/use-cases/delete-integration-for-user.js b/packages/core/integrations/use-cases/delete-integration-for-user.js index 6942882c4..378852337 100644 --- a/packages/core/integrations/use-cases/delete-integration-for-user.js +++ b/packages/core/integrations/use-cases/delete-integration-for-user.js @@ -1,3 +1,4 @@ +const { resolveIntegrationClass } = require('../integration-types'); const Boom = require('@hapi/boom'); const { getLogger } = require('../../logs'); @@ -44,10 +45,9 @@ class DeleteIntegrationForUser { ); } - const integrationClass = this.integrationClasses.find( - (integrationClass) => - integrationClass.Definition.name === - integrationRecord.config.type + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + integrationRecord.config.type ); if (integrationRecord.userId !== userId) { diff --git a/packages/core/integrations/use-cases/get-integration-for-user.js b/packages/core/integrations/use-cases/get-integration-for-user.js index f7f2caf56..52d470389 100644 --- a/packages/core/integrations/use-cases/get-integration-for-user.js +++ b/packages/core/integrations/use-cases/get-integration-for-user.js @@ -1,3 +1,4 @@ +const { resolveIntegrationClass } = require('../integration-types'); // Removed Integration wrapper - using IntegrationBase directly const { mapIntegrationClassToIntegrationDTO } = require('../utils/map-integration-dto'); const Boom = require('@hapi/boom'); @@ -47,8 +48,9 @@ class GetIntegrationForUser { throw Boom.forbidden('User does not have access to this integration'); } - const integrationClass = this.integrationClasses.find( - (integrationClass) => integrationClass.Definition.name === integrationRecord.config.type + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + integrationRecord.config.type ); const modules = []; diff --git a/packages/core/integrations/use-cases/get-integration-instance.js b/packages/core/integrations/use-cases/get-integration-instance.js index 6287db139..4cdb4e984 100644 --- a/packages/core/integrations/use-cases/get-integration-instance.js +++ b/packages/core/integrations/use-cases/get-integration-instance.js @@ -1,3 +1,4 @@ +const { resolveIntegrationClass } = require('../integration-types'); // Removed Integration wrapper - using IntegrationBase directly /** @@ -36,10 +37,9 @@ class GetIntegrationInstance { ); } - const integrationClass = this.integrationClasses.find( - (integrationClass) => - integrationClass.Definition.name === - integrationRecord.config.type + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + integrationRecord.config.type ); if (!integrationClass) { diff --git a/packages/core/integrations/use-cases/get-integrations-for-user.js b/packages/core/integrations/use-cases/get-integrations-for-user.js index 9e9efef81..846f78682 100644 --- a/packages/core/integrations/use-cases/get-integrations-for-user.js +++ b/packages/core/integrations/use-cases/get-integrations-for-user.js @@ -1,3 +1,4 @@ +const { resolveIntegrationClass } = require('../integration-types'); // Removed Integration wrapper - using IntegrationBase directly const { mapIntegrationClassToIntegrationDTO, @@ -48,10 +49,9 @@ class GetIntegrationsForUser { integrationRecord.entitiesIds ); - const integrationClass = this.integrationClasses.find( - (integrationClass) => - integrationClass.Definition.name === - integrationRecord.config.type + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + integrationRecord.config.type ); const modules = []; diff --git a/packages/core/integrations/use-cases/get-possible-integrations.js b/packages/core/integrations/use-cases/get-possible-integrations.js index 00886aa67..645d2933c 100644 --- a/packages/core/integrations/use-cases/get-possible-integrations.js +++ b/packages/core/integrations/use-cases/get-possible-integrations.js @@ -1,3 +1,5 @@ +const { listIntegrationClasses } = require('../integration-types'); + /** * Use case for retrieving all possible integration types that can be created. * @class GetPossibleIntegrations @@ -17,8 +19,11 @@ class GetPossibleIntegrations { * @async * @returns {Promise} Array of integration option details for all available integration types. */ - async execute() { - return this.integrationClasses.map((integrationClass) => + async execute({ userId } = {}) { + // Registered classes and the types providers offer (published definitions, and the ones this user's workspace + // built), one list. + const classes = await listIntegrationClasses(this.integrationClasses, userId ? { workspaceId: String(userId) } : undefined); + return classes.map((integrationClass) => integrationClass.getOptionDetails() ); } diff --git a/packages/core/integrations/use-cases/run-trigger-tick.js b/packages/core/integrations/use-cases/run-trigger-tick.js index bac8547e3..c45d829e8 100644 --- a/packages/core/integrations/use-cases/run-trigger-tick.js +++ b/packages/core/integrations/use-cases/run-trigger-tick.js @@ -6,7 +6,8 @@ * so the slot it falls in, and therefore every due decision and every `scheduledFor`, is the same if the same tick * is delivered twice or carried on by a continuation. * - * 1. Page through the ENABLED installations of this integration, `pageSize` (500) at a time, by id. + * 1. Page through the ENABLED installations of this integration, `pageSize` (500) at a time, by id. A class that hosts + * other installation types (`static triggerTypes()`, the shared runtime of published definitions) pages each. * 2. For each installation and trigger, resolve config, `when` and the cadence (integrations/triggers.js). * 3. Due: `every` when `slot mod n == hash(integrationId) mod n`; `cron` when it fired in (previous tick, this tick]; * `at` always (it is its own schedule). @@ -27,7 +28,7 @@ const MAX_DELAY_SECONDS = 900; // SQS DelaySeconds ceiling class RunTriggerTick { /** * @param {Object} deps - * @param {Function} deps.integrationClass - the IntegrationBase subclass (Definition.name, triggersFor) + * @param {Function} deps.integrationClass - the IntegrationBase subclass (Definition.name, triggersFor, optional triggerTypes) * @param {Object} deps.integrationRepository - findIntegrations({ type, status, limit, after }) * @param {Object} deps.queue - { batchSend(entries, queueUrl, { delaySeconds }), send(message, queueUrl) } * @param {string} deps.queueUrl @@ -48,7 +49,8 @@ class RunTriggerTick { * @param {string} [params.tick] - '5m': evaluate the every/cron triggers on this tick * @param {string} [params.at] - an `at` trigger's name: run it for every ENABLED installation * @param {string} [params.time] - the schedule's firing time (ISO); defaults to now - * @param {string|number} [params.cursor] - continue after this installation id + * @param {string|number|{type: string, after: string|number|null}} [params.cursor] - continue after this installation id + * (in this type, when the class covers several) * @param {() => number} [params.remainingMs] - time left in this invocation * @returns {Promise<{ installations: number, sent: number, pages: number, continued: boolean, failed: Array }>} */ @@ -67,10 +69,25 @@ class RunTriggerTick { }; const delayOf = (m) => (at ? 0 : hashId(`${m.data.integrationId}:${m.data.trigger}`) % Math.min(tickMinutes * 60, MAX_DELAY_SECONDS)); - const out = { installations: 0, sent: 0, pages: 0, continued: false, failed: [] }; + // The installation types this tick covers: the class's own, unless it hosts others (a shared runtime whose + // published definitions are integration types of their own; static triggerTypes). With more than one, the + // continuation cursor names the type it stopped in: `{ type, after }`. + const types = typeof this.Integration.triggerTypes === 'function' ? await this.Integration.triggerTypes() : [name]; + let typeIndex = 0; let after = cursor; + if (cursor && typeof cursor === 'object') { + typeIndex = Math.max(0, types.indexOf(cursor.type)); + after = cursor.after ?? null; + } + const cursorAt = (type, id) => (types.length > 1 ? { type, after: id } : id); + const continueAt = async (nextCursor) => { + await this.queue.send({ event: TRIGGER_TICK, data: { ...(at ? { at } : { tick }), time: new Date(timeMs).toISOString(), cursor: nextCursor } }, this.queueUrl); + }; + + const out = { installations: 0, sent: 0, pages: 0, continued: false, failed: [] }; for (;;) { - const rows = await this.repository.findIntegrations({ type: name, status: 'ENABLED', limit: this.pageSize, after }); + const type = types[typeIndex]; + const rows = await this.repository.findIntegrations({ type, status: 'ENABLED', limit: this.pageSize, after }); out.pages += 1; const messages = []; for (const row of rows) { @@ -95,10 +112,16 @@ class RunTriggerTick { await this.queue.batchSend(messages, this.queueUrl, { delaySeconds: delayOf }); out.sent += messages.length; } - if (rows.length < this.pageSize) return out; - after = rows[rows.length - 1].id; + if (rows.length < this.pageSize) { + // This type is done; carry on with the next one, from its start. + typeIndex += 1; + if (typeIndex >= types.length) return out; + after = null; + } else { + after = rows[rows.length - 1].id; + } if (remainingMs() < DEADLINE_MARGIN_MS) { - await this.queue.send({ event: TRIGGER_TICK, data: { ...(at ? { at } : { tick }), time: new Date(timeMs).toISOString(), cursor: after } }, this.queueUrl); + await continueAt(cursorAt(types[typeIndex], after)); out.continued = true; return out; } diff --git a/packages/core/integrations/use-cases/update-integration.js b/packages/core/integrations/use-cases/update-integration.js index c3f3bec35..6db136dae 100644 --- a/packages/core/integrations/use-cases/update-integration.js +++ b/packages/core/integrations/use-cases/update-integration.js @@ -1,3 +1,4 @@ +const { resolveIntegrationClass } = require('../integration-types'); const { mapIntegrationClassToIntegrationDTO, } = require('../utils/map-integration-dto'); @@ -41,10 +42,9 @@ class UpdateIntegration { } // 2. Get the correct Integration class by type - const integrationClass = this.integrationClasses.find( - (integrationClass) => - integrationClass.Definition.name === - integrationRecord.config.type + const integrationClass = await resolveIntegrationClass( + this.integrationClasses, + integrationRecord.config.type ); if (!integrationClass) { diff --git a/packages/core/prisma-mongodb/schema.prisma b/packages/core/prisma-mongodb/schema.prisma index 09d6f550f..34e482460 100644 --- a/packages/core/prisma-mongodb/schema.prisma +++ b/packages/core/prisma-mongodb/schema.prisma @@ -475,12 +475,15 @@ model IntegrationDefinitionVersion { previousVersion String? notes String? createdBy String? + // The workspace an integration a user built belongs to (null: the app's own catalogue, offered to everyone). + workspaceId String? publishedAt DateTime? createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@unique([definitionId, version]) @@index([definitionId, status]) + @@index([workspaceId, status]) @@map("IntegrationDefinitionVersion") } diff --git a/packages/core/prisma-postgresql/migrations/20261019120000_definition_workspace/migration.sql b/packages/core/prisma-postgresql/migrations/20261019120000_definition_workspace/migration.sql new file mode 100644 index 000000000..d7aa0864d --- /dev/null +++ b/packages/core/prisma-postgresql/migrations/20261019120000_definition_workspace/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "IntegrationDefinitionVersion" ADD COLUMN "workspaceId" TEXT; + +-- CreateIndex +CREATE INDEX "IntegrationDefinitionVersion_workspaceId_status_idx" ON "IntegrationDefinitionVersion"("workspaceId", "status"); diff --git a/packages/core/prisma-postgresql/schema.prisma b/packages/core/prisma-postgresql/schema.prisma index 2d1f78835..a2cd7401e 100644 --- a/packages/core/prisma-postgresql/schema.prisma +++ b/packages/core/prisma-postgresql/schema.prisma @@ -456,12 +456,15 @@ model IntegrationDefinitionVersion { previousVersion String? notes String? createdBy String? + // The workspace an integration a user built belongs to (null: the app's own catalogue, offered to everyone). + workspaceId String? publishedAt DateTime? createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@unique([definitionId, version]) @@index([definitionId, status]) + @@index([workspaceId, status]) } // One row per data-defined workflow run. trace holds step ids, statuses, timings and error messages, never request or diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index b61d77d6f..a8a2e2763 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -360,6 +360,25 @@ class IntegrationBuilder extends InfrastructureBuilder { method: 'POST', }, }, + // A provider (the shared runtime for published integration definitions) also receives the + // webhooks of every type it provides at that type's own URL. HTTP API cannot match part of a + // segment, so the whole segment is a parameter; literal routes still win over it. + ...(integration.IntegrationProvider + ? [ + { + httpApi: { + path: '/api/{integrationType}/webhooks', + method: 'POST', + }, + }, + { + httpApi: { + path: '/api/{integrationType}/webhooks/{integrationId}', + method: 'POST', + }, + }, + ] + : []), ], }; console.log(` ✓ Webhook handler function defined`); diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js index cc0f1c71b..e7769758d 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -620,6 +620,28 @@ describe('IntegrationBuilder', () => { ]); }); + it('gives a provider of integration types the webhook routes of every type it provides', async () => { + const appDefinition = { + integrations: [ + { Definition: { name: 'stripe', webhooks: true } }, + { + Definition: { name: 'data-defined', webhooks: { enabled: true } }, + IntegrationProvider: { resolve: async () => null, list: async () => [] }, + }, + ], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + + expect(result.functions['data-definedWebhook'].events.map((e) => e.httpApi.path)).toEqual([ + '/api/data-defined-integration/webhooks', + '/api/data-defined-integration/webhooks/{integrationId}', + '/api/{integrationType}/webhooks', + '/api/{integrationType}/webhooks/{integrationId}', + ]); + expect(result.functions.stripeWebhook.events).toHaveLength(2); + }); + it('emits a dedicated per-binding function (namespaced) for Tier 3 extension routes; the main handler keeps only {proxy+}', async () => { const appDefinition = { integrations: [