Writing Custom Connectors

Step-by-step guide to authoring, testing, and registering custom data ingestion connectors in MemoFS.

The MemoFS connector architecture is provider-neutral and extensible. Adding a new ingestion source — whether for Linear, Jira, Slack, internal wiki APIs, or custom database tables — is accomplished by implementing the Connector interface.

Like embedder and extractor adapters across MemoFS, custom connectors act as pure fetch and normalize plugins. Connectors never write files directly to disk; the runner manages deduplication, deterministic ID generation, and the single-writer write discipline.

1. The `Connector` Interface

Every connector implements the Connector interface:

import type {
  Connector,
  ConnectorIngestContext,
  ConnectorRecord,
} from "@memofs/connectors";

export interface Connector {
  /** Matches ConnectorConfig.type (e.g. "linear", "jira"). */
  readonly type: string;
  /** Human-readable display name for logs and error messages. */
  readonly displayName: string;
  /**
   * Fetch and normalize external items into ConnectorRecords.
   * Does NOT write to disk — the runner handles deduplication and storage.
   */
  ingest(ctx: ConnectorIngestContext): Promise<readonly ConnectorRecord[]>;
}

`ConnectorIngestContext`

When runConnectors() invokes connector.ingest(ctx), it provides a runtime context:

Property	Type	Description
`ctx.config`	`ConnectorConfig`	The connector's configuration row from `.memofs/connectors.json` (including `sourceMapping` and `id`).
`ctx.token`	`string`	The resolved plain-text secret token in memory. Never log or write this value to disk.
`ctx.memo`	`MemoFS`	The host's `MemoFS` instance (maintains the single-writer lock). Connectors must not instantiate their own `MemoFS` client.
`ctx.signal`	`AbortSignal?`	Optional cancellation abort signal passed from the caller.

2. Producing `ConnectorRecord` Objects

Your ingest() implementation fetches external resources and maps each item into a ConnectorRecord:

export interface ConnectorRecord {
  /** Stable external ID (e.g. "linear:ENG-104"). Used as deduplication key. */
  readonly externalId: string;
  /** One-line title for the note. */
  readonly title: string;
  /** Full markdown content of the note. */
  readonly content: string;
  /** Optional HTTP(S) URL for external provenance. */
  readonly url?: string;
  /** Optional ISO 8601 creation/occurrence timestamp. */
  readonly occurredAt?: string;
  /** Optional structured metadata dictionary for vector/search filtering. */
  readonly metadata?: JsonObject;
}

3. Complete Example: Linear Connector

Below is a complete, production-ready custom connector for Linear:

import type {
  Connector,
  ConnectorIngestContext,
  ConnectorRecord,
} from "@memofs/connectors";

interface LinearSourceMapping {
  teamKey?: string;
  limit?: number;
}

export class LinearConnector implements Connector {
  readonly type = "linear";
  readonly displayName = "Linear";

  async ingest(ctx: ConnectorIngestContext): Promise<readonly ConnectorRecord[]> {
    const mapping = (ctx.config.sourceMapping ?? {}) as LinearSourceMapping;
    const teamKey = mapping.teamKey;
    const limit = mapping.limit ?? 50;

    // Check for cancellation before network calls
    if (ctx.signal?.aborted) {
      throw new Error("Linear ingest aborted.");
    }

    // Query Linear GraphQL API with the resolved in-memory token
    const response = await fetch("https://api.linear.app/graphql", {
      method: "POST",
      headers: {
        Authorization: ctx.token,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        query: `
          query($filter: IssueFilter, $first: Int!) {
            issues(filter: $filter, first: $first) {
              nodes {
                id
                identifier
                title
                description
                url
                createdAt
                state { name }
                team { key }
              }
            }
          }
        `,
        variables: {
          first: limit,
          filter: teamKey ? { team: { key: { eq: teamKey } } } : {},
        },
      }),
      signal: ctx.signal,
    });

    if (!response.ok) {
      throw new Error(`Linear API request failed: ${response.status} ${response.statusText}`);
    }

    const payload = await response.json();
    const issues = payload.data?.issues?.nodes ?? [];

    // Normalize into standard ConnectorRecords
    return issues.map((issue: any): ConnectorRecord => {
      const description = (issue.description ?? "").slice(0, 4000);
      const content = [
        `# [${issue.identifier}] ${issue.title}`,
        "",
        description,
        "",
        `Source: ${issue.url}`,
      ].join("\n");

      return {
        externalId: `linear:${issue.identifier}`,
        title: `[${issue.identifier}] ${issue.title}`,
        content,
        url: issue.url,
        occurredAt: issue.createdAt,
        metadata: {
          team: issue.team?.key,
          state: issue.state?.name,
          identifier: issue.identifier,
        },
      };
    });
  }
}

4. Registering and Executing

import { createNodeMemoFs } from "@memofs/core/node-fs";
import {
  createConnectorRegistry,
  runConnectors,
  EnvSecretResolver,
} from "@memofs/connectors";
import { LinearConnector } from "./linear-connector";

const rootDir = ".";
const memo = createNodeMemoFs({ rootDir });

// 1. Create a registry with built-ins + custom Linear connector
const registry = createConnectorRegistry([new LinearConnector()]);

// 2. Run ingestion
const result = await runConnectors({
  rootDir,
  memo,
  secretResolver: new EnvSecretResolver({ rootDir }),
  connectorRegistry: registry,
  onlyType: "linear", // Optional filter
});

console.log("Ingestion results:", result);

import { MemoFS } from "@memofs/core";
import { createNodeFsMemoryStore } from "@memofs/core/node-fs";
import {
  createConnectorRegistry,
  runConnectors,
  EnvSecretResolver,
} from "@memofs/connectors";
import { LinearConnector } from "./linear-connector";

const rootDir = ".";
const store = createNodeFsMemoryStore({ rootDir });
const memo = new MemoFS({ store, projectId: "my-project", mode: "local" });

// 1. Create a registry with built-ins + custom Linear connector
const registry = createConnectorRegistry([new LinearConnector()]);

// 2. Run ingestion
const result = await runConnectors({
  rootDir,
  memo,
  secretResolver: new EnvSecretResolver({ rootDir }),
  connectorRegistry: registry,
  onlyType: "linear",
});

console.log("Ingestion results:", result);

Registry Inspection Methods

ConnectorRegistry provides inspection methods:

const registry = createConnectorRegistry();
registry.register(new LinearConnector());

registry.has("linear");        // true
registry.get("linear");        // LinearConnector instance
registry.types();              // ["github", "notion", "linear"]

5. Testing Custom Connectors

Because connectors only return ConnectorRecord objects without touching the filesystem, unit testing is simple:

import { describe, it, expect } from "vitest";
import { LinearConnector } from "./linear-connector";

describe("LinearConnector", () => {
  it("normalizes issues correctly", async () => {
    const connector = new LinearConnector();
    
    // Mock fetch or verify pure normalization helpers
    expect(connector.type).toBe("linear");
    expect(connector.displayName).toBe("Linear");
  });
});

6. Helper Utilities

@memofs/connectors exports helper utilities for inspecting, validating, and manipulating connector configs:

Utility	Type Signature	Description
`connectorNoteId(record)`	`(record: ConnectorRecord) => Promise<string>`	Computes the deterministic note ID (`conn_<sha256[:16]>`) without writing to disk.
`readConnectorsFile(rootDir)`	`(rootDir: string) => Promise<ConnectorsFile>`	Reads and validates `.memofs/connectors.json`. Returns `EMPTY_CONNECTORS_FILE` if missing.
`validateConnectorsFile(raw)`	`(raw: unknown) => ConnectorsFile`	Pure structural and secret-leak validator. Throws `ConnectorConfigError` on violations.
`selectConnectors(file, opts)`	`(file: ConnectorsFile, opts?: { enabled?: boolean; type?: string }) => ConnectorConfig[]`	Filters connector rows by `enabled` state and/or `type`.
`EMPTY_CONNECTORS_FILE`	`ConnectorsFile`	Frozen constant `{ connectors: [] }`.

7. Error Handling Best Practices

When authoring connectors, follow these best practices for error propagation:

Fatal Connector Errors: If an API endpoint fails (e.g. rate limit, bad credentials, network timeout), throw an Error immediately. The runner records the error under result.errors and proceeds to the next connector without writing partial records.
Cooperative Cancellation: Check ctx.signal?.aborted in loops or pass ctx.signal to native fetch().
No Secret Leaks: Never include raw tokens, passwords, or bearer headers in thrown error messages or note content.