Skip to main content

Architecture

n8n-sync is two hook bundles wired into n8n's external-hooks system. There is no server process of its own and no shared database; the publisher pushes events over HTTP and the subscriber applies them through n8n's own repositories.

publisher/index.ts  ── export = createPublisherHooks({ emit })
                      emit = fan out to one createEventSender per SYNC_SUBSCRIBER_URLS entry
                      each sender = serialized in-memory queue → sendSyncEvent
                      (fetch POST + retry + HMAC/bearer auth, never throws)
subscriber/index.ts ── export = createSubscriberHooks({ ready })
                      ready: buildN8nSyncRepositories() → createApplier()
                             → createSyncRouteHandler() → mountSyncRoutes()

Publisher side​

  • Entry file (publisher/index.ts) ends with export = createPublisherHooks(...). n8n loads hook files via require() and expects the hook map directly (IExternalHooksFileData).
  • emit is a fan-out function that enqueues one event per target URL into one createEventSender per entry in SYNC_SUBSCRIBER_URLS.
  • createEventSender (publisher/sender.ts) maintains a per-target serialized in-memory queue. Events for a given target are delivered one at a time in emitted hook order; a slow target never delays others. Exact queued duplicates of the same semantic operation may be coalesced, but mixed operations stay ordered. Hooks themselves only enqueue (fire-and-forget) so n8n stays responsive.
  • sendSyncEvent (shared/http.ts) performs a fetch POST with timeout and exponential-backoff retry (1s, 2s, 4s, capped at 10s). Every attempt re-signs the request (HMAC mode) so a retried request gets a fresh timestamp.
  • Event ordering (publisher/order-state.ts) allocates a unique source-scoped eventId and monotonic per-entity entityRevision. When delivery is enabled, SYNC_SOURCE_ID is required and persisted with the counters under SYNC_PUBLISHER_STATE_PATH.

Wired hooks → events​

n8n hookEvent
credentials.create / credentials.updatecredentials.upsert
credentials.deletecredentials.delete
workflow.afterCreate / workflow.afterUpdateworkflow.upsert
workflow.activateworkflow.activate
workflow.afterDeleteworkflow.delete
workflow.afterArchive / workflow.afterUnarchiveworkflow.archive
workflow.postExecute ★execution.upsert

★ workflow.postExecute is opt-in via SYNC_ENTITIES. See Wired Hooks.

Deliberately not wired: workflow.preExecute (fires per execution with no execution-summary counterpart on the subscriber), workflow.create/update/delete pre-hooks (redundant with the after-hooks).

SYNC_ENTITIES gating​

All SYNC_ENTITIES access lives in src/shared/config.ts as a ReadonlySet<'workflows' | 'credentials' | 'executions'>. When the env var is absent or blank, it defaults to workflows,credentials (executions are off). Explicit invalid names fail startup, and executions requires workflows.

When an entity is disabled, the corresponding hook handler is not wired at all — the key is absent from the returned hook map. n8n pays zero fan-out overhead for it. For example, with the default value the publisher emits no execution events; workflow.postExecute is re-registered only when executions is in SYNC_ENTITIES.

Subscriber side​

The subscriber mounts POST /rest/sync/v1/events plus GET …/health and GET …/ready on n8n's own server inside the n8n.ready hook, and applies events via n8n's internal repositories.

  • n8n.ready is the only hook the subscriber wires. It resolves n8n's DI Container and from it pulls only the repository services needed for enabled entity families. Resolving DI earlier crashes; nothing in the bundle touches the container before ready fires.
  • createApplier(repos, opts) (subscriber/applier.ts) is the heart of the subscriber. It enforces source-scoped revision ordering before repository writes, then uses row timestamp guards for last-write-wins behavior.
  • createSyncRouteHandler + mountSyncRoutes (subscriber/routes.ts) wire the HTTP entry point. The flow per request is: readiness → authenticate → validate → apply → 200 { "ok": true }.

Wire format​

The wire payload is the SyncEvent discriminated union in src/shared/types.ts. Inbound payloads are validated with parseSyncEvent (src/shared/validate.ts); anything that does not match the contract is rejected with a 400 before any repository write happens.

Idempotency and ordering​

Ordering has two layers:

  1. Source-scoped entityRevision ordering is persisted under SYNC_SUBSCRIBER_STATE_PATH, including delete tombstones.
  2. Accepted upserts then use a row timestamp guard to avoid regressing target rows.
EntityInvariant columnNote
WorkflowsupdatedAtStored row is skipped when its updatedAt ≥ incoming.
CredentialsupdatedAtSame guard.
ExecutionsstoppedAtIn-flight executions (stoppedAt: null) skip the guard so a later delivery can still converge state.

Exact duplicate deliveries with the same eventId and entityRevision are no-ops. A distinct eventId that reuses an already-applied entityRevision returns 409 SYNC_REVISION_CONFLICT and does not modify the entity or checkpoint. After revision ordering accepts a newer event, equal timestamps may update the row; older timestamps are rejected without advancing the checkpoint.

Deletes and archives for unknown ids are no-ops. Workflow deletion also removes mapped synced execution rows for that source/workflow before deleting the workflow.

Conventions​

  • All process.env access lives in src/shared/config.ts — nowhere else in the bundle.
  • n8n payload types (IWorkflowBase, ICredentialsDb, N8nServer) are local minimal copies in src/shared/types.ts. The package must stay free of n8n dependencies.
  • The logger is the zero-dep structured JSON logger in src/shared/logger.ts (createLogger(module)).
  • The subscriber uses n8n's global rawBodyReader middleware, reading request bodies from req.rawBody when available, with a zero-dep stream read fallback. HMAC mode fails closed if only a pre-parsed req.body remains, because verification must use the exact raw bytes that are parsed and applied.

Before and after​

// polling script on the target, runs every minute
const workflows = await fetch('https://source.example.com/api/v1/workflows', {
  headers: { 'X-N8N-API-KEY': process.env.SOURCE_API_KEY },
}).then(r => r.json());

for (const wf of workflows.data) {
  const existing = await localDb.workflow.findOne({ where: { id: wf.id } });
  if (!existing || new Date(wf.updatedAt) > new Date(existing.updatedAt)) {
    await localDb.workflow.save(wf);
  }
}