@memberjunction/integration-engine

MemberJunction Integration Engine — orchestration, field mapping, and connector framework for synchronizing external systems with MJ entities.

Architecture

The integration engine follows a pipeline architecture:

External System → Connector → FieldMappingEngine → MatchEngine → MJ Entity Persistence
Copy

Core Components

Field mapping — the shared transform engine

FieldMappingEngine owns only the integration-specific concerns: source flattening, field-map iteration, and unmapped-field overflow capture. The actual per-value transform pipeline (direct / regex / split / combine / lookup / format / coerce / substring / custom) is not implemented here — it's the shared FieldTransformEngine in @memberjunction/global, so integration sync and the rules-based bulk-update tool run the exact same transform implementation (fix or extend a transform once, everywhere benefits).

Integration is the right home when the other side is a live external system — it owns the protocol, auth, discovery, match resolution, sync direction, watermarks, and dead-lettering, and borrows the shared engine only for the field transforms. When both sides are MJ data (updating an entity from its own fields / related entities / provided context), use EntityFieldRules in @memberjunction/core instead — same engine underneath, metadata-aware on top.

Quick Start

import { IntegrationOrchestrator } from '@memberjunction/integration-engine';

const orchestrator = new IntegrationOrchestrator();
const result = await orchestrator.RunSync(companyIntegrationID, contextUser, 'Manual');

console.log(`Processed: ${result.RecordsProcessed}`);
console.log(`Created: ${result.RecordsCreated}, Updated: ${result.RecordsUpdated}`);
console.log(`Errors: ${result.RecordsErrored}`);
Copy

Creating a Custom Connector

import { RegisterClass } from '@memberjunction/global';
import {
  BaseIntegrationConnector,
  ConnectionTestResult,
  ExternalObjectSchema,
  ExternalFieldSchema,
  FetchContext,
  FetchBatchResult,
} from '@memberjunction/integration-engine';

@RegisterClass(BaseIntegrationConnector, 'HubSpotConnector')
export class HubSpotConnector extends BaseIntegrationConnector {
  async TestConnection(companyIntegration, contextUser): Promise<ConnectionTestResult> {
    // Test HubSpot API connectivity
  }

  async DiscoverObjects(companyIntegration, contextUser): Promise<ExternalObjectSchema[]> {
    // Return available HubSpot objects (Contacts, Companies, Deals, etc.)
  }

  async DiscoverFields(companyIntegration, objectName, contextUser): Promise<ExternalFieldSchema[]> {
    // Return fields for a specific HubSpot object
  }

  async FetchChanges(ctx: FetchContext): Promise<FetchBatchResult> {
    // Fetch changed records using HubSpot's API with watermark-based pagination
  }
}
Copy

Transform Types

The FieldMappingEngine supports 9 transform types configured via JSON in the TransformPipeline field:

Transform Pipeline Example

[
  { "Type": "regex", "Config": { "Pattern": "[^0-9]", "Replacement": "", "Flags": "g" } },
  { "Type": "substring", "Config": { "Start": 0, "Length": 3 } }
]
Copy

This pipeline strips non-numeric characters then extracts the first 3 digits (e.g., area code from phone number).

Error Handling

Each transform step can specify OnError:

• "Fail" (default) — throws, halting the record
• "Skip" — skips the field entirely
• "Null" — sets the field to null

API Reference

Notifications

After a sync completes or fails, an optional onNotification callback receives a SyncNotification with pre-formatted subject and body text suitable for email delivery:

const result = await orchestrator.RunSync(
    companyIntegrationID, contextUser, 'Scheduled',
    undefined,  // onProgress
    (notification) => {
        if (notification.Severity !== 'Info') {
            emailService.send({
                to: 'ops@example.com',
                subject: notification.Subject,
                text: notification.Body,
            });
        }
    }
);
Copy

SyncNotification properties:

• Event: 'SyncCompleted' | 'SyncCompletedWithErrors' | 'SyncFailed'
• Severity: 'Info' | 'Warning' | 'Error'
• Subject / Body: human-readable text ready for email
• Result: the full SyncResult for programmatic access

IntegrationOrchestrator

• RunSync(companyIntegrationID, contextUser, triggerType?, onProgress?, onNotification?) — Executes a full sync run

FieldMappingEngine

• Apply(records, fieldMaps, entityName) — Maps external records to MJ entity fields

MatchEngine

• Resolve(records, entityMap, fieldMaps, contextUser) — Resolves Create/Update/Delete/Skip

WatermarkService

• Load(entityMapID, contextUser) — Loads current watermark
• Update(entityMapID, newValue, contextUser) — Updates or creates watermark

ConnectorFactory

• Resolve(integration, sourceTypes) — Creates connector instance via ClassFactory

BaseIntegrationConnector (abstract)

• TestConnection(companyIntegration, contextUser) — Tests external system connectivity
• DiscoverObjects(companyIntegration, contextUser) — Lists available external objects
• DiscoverFields(companyIntegration, objectName, contextUser) — Lists fields on an object
• FetchChanges(ctx) — Fetches a batch of changed records
• GetDefaultFieldMappings(objectName, entityName) — Suggests default mappings

Environment Variables

Operator-tunable knobs read from the process environment. All are optional with safe defaults.

Schema materialization caps (operator guardrails)

These bound what a create-tables / RSU apply (IntegrationApplyAll / ApplyAllBatch / ApplySchemaBatch) may materialize. They are deliberately env-only — NOT per-connection or GraphQL-settable — because a cap a user could raise via the same API they apply with would be no guardrail. An over-limit selection is rejected with a clear error (never truncated); discovery still surfaces every object/field, only materialization is capped.

Discovery (stage-2 streaming field discovery — no-describe / file-feed sources only)

Bound the data sample used to build a column corpus + lightweight PK guess (NOT a full scan). Also overridable per-connection via IntegrationSetSyncConfig (discoveryBatchSize / discoveryMaxRecords / discoveryTimeBudgetMs); precedence is explicit-opts > per-connection Configuration > env > default.

Sync / runtime