DEV Community: Abdulkadir N. A.

Data synchronization with state machines (XState) and data-loaders in NodeJS

Abdulkadir N. A. — Wed, 10 Jan 2024 14:41:23 +0000

How would you build out a solution to the following problem? You receive a stream containing several hundred to many thousand rows of records, each potentially corresponding to an entity in your system(s), where each row needs to be processed individually to determine what combination of 0 to many successive operations both internally and externally to be performed per record in those systems/services in a generic fashion that's scalable, extensible, performant, auditable, interruptible, idempotent, testable, respects rate limits and is relatively simple to understand. Oh, and this solution needs to describe every single operation it'll perform to the operator for the batch for green lighting (👍🏾 / 👎🏾) before kicking off the planned steps, and also return a post-operation report which in most cases, should match the pre-run report. Stick with me as I describe how I built this out for some of my clients in the past.

The TLDR in my case, was a combination of (small) State machines, Data loaders, and a lot of plumbing. The state machines are for logic determination, coordination, and keeping things small and simple, while the data loaders are for batching operations so that the whole thing isn't awfully slow because it performs each operation 1 by 1.

More concretely, a row above could describe an entity from a file (in one client's domain in the past) that needs to conditionally get created in a DB, updated if it exists, update an eligibility status, or soft deleted to de-activate a user, created in a CRM, updated in the CRM if the record exists, created, read, updated, or deleted all conditionally in an LMS, entitlement system, authentication service, search cluster or any other services/systems that should have the representation of the row etc. Other use cases include a chat message that needs to be scrubbed for PII, deleted from a store, indexed in a search DB, and updated stats.

The problem boils down to the first principles for each record, perform some checks and then make a decision on the action to take given the state of the entity, then perform the operation or operations, rinse and repeat while handling failures (transient and non-transient), and being interruptible, cancellable, fast, introspect-able and more.

Actions to each service can be modeled with the Typescript types below using Database (DB), Salesforce, Active Campaign, and Elastic Search as example services for this post.

type DBAction = "NOTHING" | "CREATE" | "UPDATE" | "DELETE";

type SalesforceAction =
  | "NOTHING"
  | "CREATE_CONTACT"
  | "UPDATE_CONTACT"
  | "DELETE_CONTACT"
  | "ACTIVATE_CONTACT"
  | "DEACTIVATE_CONTACT";

type ActiveCampaignAction =
  | "NOTHING"
  | "CREATE_CONTACT"
  | "UPDATE_CONTACT"
  | "DELETE_CONTACT"
  | "ACTIVATE_CONTACT"
  | "DEACTIVATE_CONTACT";

type Auth0Action =
  | "NOTHING"
  | "CREATE_USER"
  | "UPDATE_USER"
  | "DELETE_USER"
  | "ACTIVATE_USER"
  | "DEACTIVATE_USER";

type ElasticSearchAction =
  | "NOTHING"
  | "CREATE_USER"
  | "UPDATE_USER"
  | "DELETE_USER"
  | "ACTIVATE_USER"
  | "DEACTIVATE_USER";

type ServiceAction =
  | DBAction
  | SalesforceAction
  | ActiveCampaignAction
  | Auth0Action
  | ElasticSearchAction;

As I mentioned above, the solution I implemented uses state machines and data loaders. For libraries, I use the popular XState and GraphQL Data Loader libraries. Quotes from the library documentation do a much better job of explaining what they do than I'll be able to in this post. There are also lots of excellent resources out there to gain a better understanding of these tools.

XState is a library for creating, interpreting, and executing finite state machines and state charts, as well as managing invocations of those machines as actors.

A finite state machine (FSM) is a mathematical model of computation that describes the behavior of a system that can be in only one state at any given time. State charts are a formalism for modeling stateful, reactive systems.

On the other hand,

DataLoader is a generic utility to be used as part of your application's data fetching layer to provide a simplified and consistent API over various remote data sources such as databases or web services via batching and caching.

The key thinking behind the ideas in the post is that every record can be processed "independently". The state machine is the brain (🧠) while the data loaders are the brawn (💪🏾).

In my implementation, a state machine is modeled for every record. Every row has a state machine instantiated for it. The state machine begins by interrogating the initial state of a record and figuring out what to do next based on simple conditions using invoked callback actors (a concept in XState). Transitioning into the state can create and start an actor whose logic is specified by its src property value. The onDone array can be a list of guards which is a state target node and predicate to transition to when the condition in the predicate evaluates to true.

type Mode = "DryRun" | "Live";

type GetDBActionDoneInvokeEventData = {
  module: "db";
  index: number;
  action: DBAction;
};
type GetDBActionDoneInvokeEvent =
  DoneInvokeEvent<GetDBActionDoneInvokeEventData>;

createMachine({
  // other machine configuration
  initial: "init",

  schema: {} as {
    context: {
      index: number;
      mode: Mode;

      user: AuthUser;

      input: BatchRecord;
      record?: AppRecord;

      dataLoaders: typeof loaders;
    };
  },

  states: {
    init: {
      invoke: {
        src: async (context) => {
          // load the record from the DB if it exists
          const record = await context.dataLoaders.db.user.find.load(
            context.input.email,
          );

          const params = {
            module: "db",
            index: context.index,
          } as const;

          switch (true) {
            // the record is not in the DB
            case !record:
              return Promise.resolve<GetDBActionDoneInvokeEventData>({
                ...params,
                action: "CREATE",
              });

            // the record is in the DB but the active flags don't match
            case context.input.active !== record?.active:
              return Promise.resolve<GetDBActionDoneInvokeEventData>({
                ...params,
                action: "UPDATE",
              });

            // TODO: handle other cases according to your business logic and return and action

            default:
              return Promise.resolve<GetDBActionDoneInvokeEventData>({
                ...params,
                action: "NOTHING",
              });
          }
        },
        onDone: [
          {
            cond: (c) => c.mode === "DryRun",
            target: "provisioning",
          },
          {
            cond: (_, e: GetDBActionDoneInvokeEvent) =>
              e.data.action === "CREATE",
            target: "creatingDBRecord",
          },
          {
            cond: (_, e: GetDBActionDoneInvokeEvent) =>
              e.data.action === "UPDATE",
            target: "updatingDBRecord",
          },
          {
            cond: (_, e: GetDBActionDoneInvokeEvent) =>
              e.data.action === "DELETE",
            target: "deletingDBRecord",
          },
        ]
      }

      // other states
      complete: {}
    }
  }
};

When the machine is interpreted with the row from the file as context, the initial state is set to the init state. Entering the state begins the invocation of the src callback. In the callback, the context for the row is loaded from the database using a data loader. If the record does not exist, an action (CREATE) is returned from the callback. If say some expected flags or fields don't match, other actions (UPDATE, DELETE, or DEACTIVATE) are returned from the callback. If there's nothing to be done, the NOTHING action is returned.

Next, onDone defines the transition to be taken based on the action returned from the src callback. Remember one of the requirements I described above was that the system needs to describe the sequence of actions to be taken for auditing and approval by an operator before mutation? The first element in the onDone array handles the condition where the machine is being interpreted in Dry Run mode. In Dry Run mode, the machine bails early stopping the machine from mutating any records, and goes to the next state called the provisioning state. If not in the dry run mode, and an action was returned, one of the conditions in onDone matches, and the correct state is transitioned to. Example creatingDBRecord if the action from src was CREATE.

creatingDBRecord: {
  invoke: {
    src: (context) => {
      // prepare create payload
      const payload = {}

      // create the record in the DB
      return context.dataLoaders.db.user.create.load(payload);
    },
    onDone: {
      target: "provisioning",
    },
  },
},
// updatingDBRecord
// deletingDBRecord

The provisioning state is transitioned to in all cases in the machine definition. The provisioning state is a parallel state node that contains children states that will get run at the same time. This is where CRUD-ing and other actions to external systems should happen Auth, CRM, Search, etc. The states are similar to the init I described above in the sense that a check is performed using a data loader for what state the record in Auth0, Salesforce, or Elasticsearch is, an action is returned, and based on the action, the correct state is transitioned to for performing the correct logic in the service/system. Using a parallel state gives us the benefit of allowing the CRUD operations to the 3rd party services to happen concurrently advancing the synchronization as fast as possible.

provisioning: {
  type: "parallel",
  onDone: "complete",

  states: {
    activeCampaign: {
      states: {
        checking: {
          invoke: {
            src: (context) => {
              // check if the record exists in activeCampaign
              const record =
                await context.dataLoaders.db.activeCampaign.contact.find.load(
                  context.input.email,
                );

              const params = {
                module: "activeCampaign",
                index: context.index,
              } as const;

              switch (true) {
                // the record is not in the DB
                case !record:
                  return Promise.resolve<GetActiveCampaignActionDoneInvokeEventData>(
                    {
                      ...params,
                      action: "CREATE_CONTACT",
                    },
                  );

                // the record is in the DB but the active flags don't match
                case context.input.active !== context.record?.active:
                  return Promise.resolve<GetActiveCampaignActionDoneInvokeEventData>(
                    {
                      ...params,
                      action: "UPDATE_CONTACT",
                    },
                  );

                default:
                  return Promise.resolve<GetActiveCampaignActionDoneInvokeEventData>(
                    {
                      ...params,
                      action: "NOTHING",
                    },
                  );
              }
            },
            onDone: [
              {
                cond: (c) => c.mode === "DryRun",
                target: "success",
              },
              {
                cond: (_, e: GetActiveCampaignActionDoneInvokeEvent) =>
                  e.data.action === "CREATE_CONTACT",
                target: "creatingContact",
              },
              {
                cond: (_, e: GetActiveCampaignActionDoneInvokeEvent) =>
                  e.data.action === "UPDATE_CONTACT",
                target: "updatingContact",
              },
              {
                cond: (_, e: GetActiveCampaignActionDoneInvokeEvent) =>
                  e.data.action === "DELETE_CONTACT",
                target: "deletingContact",
              },
              {
                cond: (_, e: GetActiveCampaignActionDoneInvokeEvent) =>
                  e.data.action === "ACTIVATE_CONTACT",
                target: "activatingContact",
              },
              {
                cond: (_, e: GetActiveCampaignActionDoneInvokeEvent) =>
                  e.data.action === "DEACTIVATE_CONTACT",
                target: "deactivatingContact",
              },
            ],
          },
        },
        creatingContact: {},
        updatingContact: {},
        activatingContact: {},
        deactivatingContact: {},
        deletingContact: {},
        success: {
          type: "final"
        },
      }
    },
    auth0: {
      // similar to activeCampaign but specialized in biz logic
    },
    elasticsearch: {
      // similar to activeCampaign but specialized in biz logic
    },
    salesforce: {
      // similar to activeCampaign but specialized in biz logic
    }
  }
}

The skeleton for the data loaders object looks like the following:

type BatchRecord = { email: string; active: boolean, ... };
type AppRecord = { id: string; email: string; active: boolean, ... };
const loaders = {
  db: {
    user: {
      find: new DataLoader((keys) => Promise.resolve(keys)),
      create: new DataLoader((keys) => Promise.resolve(keys)),
      update: new DataLoader((keys) => Promise.resolve(keys)),
      delete: new DataLoader((keys) => Promise.resolve(keys)),
    },
    activeCampaign: {
      contact: {
        find: new DataLoader((keys) => Promise.resolve(keys)),
        create: new DataLoader((keys) => Promise.resolve(keys)),
        update: new DataLoader((keys) => Promise.resolve(keys)),
        delete: new DataLoader((keys) => Promise.resolve(keys)),
      },
    },
    auth0: {
      user: {
        // similar to above
      }
    }
    salesforce: {
      contact: {
        // similar to above
      },
    },
  },
  activeCampaign: {
    contact: {
      find: new DataLoader((keys) => Promise.resolve(keys)),
      create: new DataLoader((keys) => Promise.resolve(keys)),
      update: new DataLoader((keys) => Promise.resolve(keys)),
      delete: new DataLoader((keys) => Promise.resolve(keys)),
      activate: new DataLoader((keys) => Promise.resolve(keys)),
      deactivate: new DataLoader((keys) => Promise.resolve(keys)),
    },
  },
  auth0: {
    // similar to above
  },
  salesforce: {
    // similar to above
  },
};

Active Campaign, Auth0, and Salesforce keys seem like they are duplicated in the db path and also as top-level keys. For one implementation, I stored the identifiers that our DB records point to in AC, Auth0, and Salesforce as rows in our database. Aside from having a user table, there's also a user_active_campaign_contact table that has a foreign key to the user record that holds the identifier of the user in AC. Other tables exist for every other external service. The data loader at loaders.db.activeCampaign.contact CRUDs only that identifier. The loader at loaders.activeCampaign.contact talks to the actual service over SDK, HTTP, GraphQL, or whatever communication protocol that exists to CRUD records to it.

At this point, if you haven't realized it, I want to explain what Data loaders help us achieve here. The state machines I've described above deal with individual records and know nothing about batches. It helps to keep IMO keep logic much simpler. However, if we tried to read and write database entries or talk to 3rd party services directly in the state machines, we'd encounter the famed N+1 select problems, where N state machine callbacks try to perform a select or update by themselves. That's highly un-optimal and slow. By performing the CRUD through the data loaders, we achieve the benefit of batching. N state machine callbacks call .load({…}), those get coalesced into the data loader batch function which can use a batch optimized API like SELECT…IN, Salesforce Batch calls, or Auth0 batch calls, or your 3rd party service's bulk CRUD API. The data loader is where I'll include code to handle failures, retries, logging, and any other resilience concerns. The separation in my experience has significantly improved the maintainability of this approach.

Running

To bring this all together, a simple runner script is shown here:

type ServiceActionDoneInvokeEventData =
  | GetDBActionDoneInvokeEventData
  | GetSalesforceActionDoneInvokeEventData
  | GetActiveCampaignActionDoneInvokeEventData
  | GetAuth0ActionDoneInvokeEventData;

async function run() {
  const batchSize = 50;
  const records: BatchRecord[] = [
    { email: "1@test.com", active: true },
    { email: "2@test.com", active: true },
  ];

  const actions: ServiceActionDoneInvokeEventData[][] = [];

  const pooledMachines = new Array(Math.min(batchSize, records.length))
    .fill(0)
    .map(() => {
      const interpreter = interpret(rootMachine, {
        deferEvents: false,
        devTools: false,
      });

      interpreter.onTransition((state, event) => {
        if ("data" in event) {
          const actionEvent = event as
            | GetDBActionDoneInvokeEvent
            | GetActiveCampaignActionDoneInvokeEvent
            | GetAuth0ActionDoneInvokeEvent
            | GetSalesforceActionDoneInvokeEvent;

          if (!actions[actionEvent.data.index]) {
            actions[actionEvent.data.index] = [];
          }

          console.log(actionEvent.data);
          actions[actionEvent.data.index].push(actionEvent.data);
        }
      });

      return interpreter;
    });

  let i = 0;
  for await (const batch of chunk(records, batchSize)) {
    const runPromises = pooledMachines.map((m) => {
      m.start({
        ...m.initialState,
        context: {
          index: i,
          dataLoaders: loaders,
          mode: "DryRun", // 
          user: {
            id: "123",
          },
          input: batch[i],
        },
      });

      i++;
    });

    await Promise.all(runPromises);
  }
}

run();

I create a set of pooled machines whose size is determined by various factors including your systems resources, batch limits of your 3rd party service, etc. The pooled machines have an onTransition callback defined that adds the actions to be taken by every state node into an actions array. This can be parsed into a report that's displayed to an operator to describe every action to be taken say in a Dry Run mode or a post-operation report. The records from the stream are batched and used to start each machine.

A lot more can be done in the runner.

Initializing the state machine with a custom state for start/pause/resume and debugging purposes.
Persisting state to persistent storage on transition for auditing purposes.
Logging and instrumentation.
Communicating status information.

One point to also be made is the fact that this can all be visualized using XState's visual editor (in VSCode or in the browser) or the Stately editor. Even better, the machine can be simulated in the editors to catch bugs early.

Summary

In this article, I describe a robust solution for efficiently handling streams of records using state machines and data loaders. The goal is to achieve scalability, performance, and auditability in a generic and understandable fashion.

Key Components:

State Machines and Data Loaders:
- We leverage XState for managing state machines and facilitating logic determination and coordination.
- GraphQL Data Loader optimizes data fetching by providing a simplified API over various data sources, avoiding performance bottlenecks.
Clear Action Modeling:
- TypeScript types define actions for different services, ensuring a comprehensive understanding of permissible operations.
Batching for Performance:
- Data loaders are crucial for optimizing performance by batching operations, mitigating the N+1 select problem.
Handling Different Services:
- Parallel state nodes efficiently process actions related to various services concurrently.
Operational Transparency:
- Describing the sequence of actions for auditing and approval by an operator before mutation, enhancing transparency.

Areas of Improvement:

Considerations for error handling, state persistence for auditing, scalability, testing strategies, and security are vital for a comprehensive solution.

Comment if you'd like to see a (yet-to-be-built) playground. Complete code is available here: https://gist.github.com/kadeer/5f35fd4048249b967e13bbbd4085d031

Hire me: I’m a full-stack engineer with over 10 years of experience in React, Javascript/Typescript, C#, PHP, Python, Rest/GraphQL, XState, SQL (Postgres/MySQL), AWS, and more.
LinkedIn: https://www.linkedin.com/in/abdulkadirna/
Twitter: https://twitter.com/adulkadirna

GraphQL directives for supercharging your API

Abdulkadir N. A. — Thu, 07 Dec 2023 13:13:31 +0000

Most GraphQL schemas have types representing things that need to be queried, such as Books, Items, Authors, Customers, etc. To constrain on the items returned by the queries, there are multiple approaches I’ve seen in the wild to specify filters:

# Option 1 - 🤮 filter yourself, damn it!
query Books 

# Option 2 - 🤕 What about Id and Name???
query BooksById {}
query BooksByName {}
query BooksByAuthor {}

# Option 3 - 🥺 can I haz more?
query Books(filter: {id: String, name: String}) {}

# Option 4 - ✨ awesome!
query Books(filter: {id: String, id_contains, id_in: [String]})

Option 2 is often too verbose, non-composable can be downright ugly. Option 3 starts to provide the right level of composability but wanting anything other than an equality comparison is to be left wanting. I like option 4 a lot because depending on the effort the API owners are willing to take on, provides the most compact, composable, and efficient API. For the rest of this post, I’ll be focused on Option 4.

Writing out the definitions for option 4 can be cumbersome, produce verbose schemas, is prone to errors/typos, mistakes that might be hard to catch, and are subject to variations by different members or teams. Where there are 10 entity fields and an average of 4 operators per entity field, that’s 40 fields necessary per entity multiplied by the number of entities in the schema 😱.

query Books(filter: BookFilter): [Book!]
query Authors(filter: AuthorFilter): [Book!]

type Book: {
  id: ID!
}

type Author: {
  id: ID!
}

input BookFilter {
  AND: [BookFilter!];
  OR: [BookFilter!];
  NOT: [BookFilter!];

  id: ID;

  name: String;
  name_contains: String;
  name_in: [String];
  name_startsWith: String;
  name_endsWith: String;

  quantity: Int;
  quantity_lt: Int;
  quantity_lte: Int;
  quantity_gt: Int;
  quantity_gte: Int;

  publishDate: Date;
  publishDate_lt: Date
  publishDate_lte: Date
  publishDate_gt: Date
  publishDate_gte: Date

  authors_some: AuthorFilter
  authors_every: AuthorFilter
  authors_none: AuthorFilter

  # ..other fields
}

input AuthorFilter {AND: [BookFilter!];
  OR: [BookFilter!];
  NOT: [BookFilter!];

  id: ID;

  name: String;
  name_contains: String;
  name_in: [String];
  name_startsWith: String;
  name_endsWith: String;

  # ..other fields
}

I propose a simpler approach for arriving at the same output while keeping the schema DRY-er, smaller and hopefully more manageable across team members and teams using GraphQL directives.

Directives

A directive decorates part of a GraphQL schema or operation with additional configuration. Tools like Apollo Server (and Apollo Client) can read a GraphQL document’s directives and perform custom logic as appropriate.
Directives are preceded by the @ character, like so:
https://www.apollographql.com/docs/apollo-server/schema/directives/

The core GraphQL specification includes exactly two directives, which must be supported by any spec-compliant GraphQL server implementation:
@include(if: Boolean) Only include this field in the result if the argument is true.
@skip(if: Boolean) Skip this field if the argument is true.
Directives can be useful to get out of situations where you otherwise would need to do string manipulation to add and remove fields in your query. Server implementations may also add experimental features by defining completely new directives.
https://graphql.org/learn/queries/#directives

Solution

Using directives, the schema definition shrinks significantly to

query Books(filter: BookFilter): [Book!]
query Authors(filter: AuthorFilter): [Author!]

type Book: {
  id: ID!
}

type Author: {
  id: ID!
}

input BookFilter {
  id: ID @filter
  name: String @filter
  quantity: Int @filter
  authors: AuthorFilter @filter(relation: One)
  # other fields
}

input AuthorFilter {
  id: ID @filter
  name: String @filter
  # other fields
}

To expand the schema into the complete output from earlier, we use a directive transformer. A directive called filter here is defined in the schema file. This directive can only be applied to GraphQL input fields hence the constraint.

directive @filter(relation: RelationTypes) on INPUT_FIELD_DEFINITION

enum RelationTypes {
  One
  Many
}

For the actual transformer (filterDirectiveTransformer in code below), we map over the schema using the mapSchema function from @graphql-tools/utils.

mapSchema is a powerful tool, in that it creates a new copy of the original schema, transforms GraphQL objects as specified, and then rewires the entire schema such that all GraphQL objects that refer to other GraphQL objects correctly point to the new set.

For the things we are interested in, we have registered callbacks. The first two MapperKind.DIRECTIVE and MapperKind.ENUM_TYPE removes the @filter directive, and Enum type called RelationTypes that we added to the schema definition above from the final schema so consumers of our API has no access to them. MapperKind.INPUT_OBJECT_TYPE starts by checking every input for fields with the @filter directives. There’s a MapperKind.INPUT_OBJECT_FIELD for each input field but in my case, I needed to mutate the entire input object definition, to add the AND, OR, and NOT fields and it made more sense to collect the logic all in one handler rather than track state in closures and spread the logic around.

export const SupportedScalarTypes = [
  "String",
  "Int",
  "Float",
  "Boolean",
  "Date",
  "DateTime",
] as const;

export type Relation = "One" | "Many";

const filterDirectiveTransformer = (
  schema: GraphQLSchema,
  directiveName: string,
) => {
  return mapSchema(schema, {
    [MapperKind.DIRECTIVE]: (directiveConfig) =>
      directiveConfig.name === directiveName ? null : directiveConfig,

    [MapperKind.ENUM_TYPE]: (type) =>
      type.name === "RelationTypes" ? null : type,

    [MapperKind.INPUT_OBJECT_TYPE]: (type) => {
      const fields = type.getFields();

      // get all fields with the filter directive
      const fieldsWithDirectives = Object.values(fields).filter((field) =>
        getDirective(schema, field, directiveName),
      );

      // if there are no fields with the filter directive, return early 
      if (fieldsWithDirectives.length === 0) {
        return type;
      }

      const filterFields: string[] = [];

      const relationInputDummyTypes = [];

      for (const field of fieldsWithDirectives) {
        const filterDirective = getDirective(schema, field, directiveName);
        if (filterDirective) {
          const args = (filterDirective as FilterDirectiveArgs)[0];

          // @ts-ignore
          const astType = field.astNode.type as NamedTypeNode;

          const isScalar = SupportedScalarTypes.includes(
            astType.name.value as any,
          );

          const isRelation = !!args.relation;

          if (isScalar) {
            filterFields.push(
              buildScalarFilterFields(astType.name.value as any, field.name),
            );
          }

          if (isRelation) {
            let relatedType = field.type as GraphQLInputObjectType;

            filterFields.push(
              buildRelationFilterFields(
                // @ts-ignore
                args.relation,
                field.name,
                relatedType.name,
              ),
            );

            relationInputDummyTypes.push(relatedType);
          }
        }
      }

      const source = `
        scalar Date
        scalar DateTime

        ${relationInputDummyTypes
          .filter((relatedType) => relatedType.name !== type.name)
          .map((type) => `input ${type.name} { id: ID! }`)
          .join("\n")}

        input ${type.name} {
            AND: [${type.name}!]
            OR: [${type.name}!]

            ${filterFields.join("")}
        }
      `;

      const newInputSchema = buildSchema(source);

      return new GraphQLInputObjectType({
        ...type,
        fields: {
          ...(
            newInputSchema.getType(type.name) as GraphQLInputObjectType
          ).getFields(),
        },
      });
    },
  });
};

For each field found on the Input type passed to the visitor callback, we grab its args and figure out if the field type is a supported scalar type. If the field is a scalar type, we call the buildScalarFilterFields builder to return a string based on a template. The buildRelationFilterFields function is called in the case the filter specifies a relation arg on the directive.

export const buildScalarFilterFields = (
  fieldType: (typeof SupportedScalarTypes)[number],
  fieldName: string,
) => {
  switch (fieldType) {
    case "String":
      return `
        ${fieldName}: String # equals
        ${fieldName}_not: String # not equals
        ${fieldName}_in: [String] # in
        ${fieldName}_not_in: [String] # not in
        ${fieldName}_contains: String # contains
        ${fieldName}_not_contains: String # not contains
        ${fieldName}_startsWith: String # starts with
        ${fieldName}_not_startsWith: String # not starts with
        ${fieldName}_endsWith: String # ends with
        ${fieldName}_not_endsWith: String # not ends with
      `;

    case "Int":
    case "Float":
    case "Date":
    case "DateTime":
      return `
        ${fieldName}: ${fieldType} # equals
        ${fieldName}_not: ${fieldType} # not equals
        ${fieldName}_in: [${fieldType}] # in
        ${fieldName}_not_in: [${fieldType}] # not in
        ${fieldName}_lt: ${fieldType} # less than
        ${fieldName}_lte: ${fieldType} # less than or equal
        ${fieldName}_gt: ${fieldType} # greater than
        ${fieldName}_gte: ${fieldType} # greater than or equal
      `;

    case "Boolean":
      return `
        ${fieldName}: Boolean # equals
        ${fieldName}_not: Boolean # not equals
      `;
  }
};

export const buildRelationFilterFields = (
  relationType: Relation,
  fieldName: string,
  inputTypeName: string,
) => {
  switch (relationType) {
    case "One":
      return `
          ${fieldName}_is: ${inputTypeName} # condition must be true for all
          ${fieldName}_is_not: ${inputTypeName} # condition must be false for all
        `;
    case "Many":
      return `
          ${fieldName}_every: [${inputTypeName}] # condition must be true for all
          ${fieldName}_some: [${inputTypeName}] # condition must be true for at least one
          ${fieldName}_none: [${inputTypeName}] # condition must be false for all
      `;
  }
};

In the transformer, we combine the filters into a new type definition along with “dummy” types we collect that are relations into a string that’s parsed by buildSchema from graphql/utilities. Dummy types here are other input types that are referenced in the input type being transformed. For example, property author in BookFilter might be of type AuthorFilter in the schema. Calling buildSchema without a dummy AuthorFilter when transforming the BookFilter throws an error due to the missing type. Finally, we collected all the fields on the newly built schema for the filter fields and return a new GraphQLInputObjectType instance.

The transformer can be applied to a schema and tested against Apollo Server with the following:

let schema = makeExecutableSchema({
  typeDefs: readFileSync("./src/schema/schema.graphql", "utf8"),
  resolvers,
});

const server = new ApolloServer({
  schema: filterDirectiveTransformer(schema, "filter"),
});

async function run() {
  const { url } = await startStandaloneServer(server, {
    listen: { port: 4000 },
  });

  console.log(`🚀  Server ready at: ${url}`);
}

run();

Here’s a screenshot of the output schema in Apollo Studio after the transformer has been applied to a schema similar to the one defined above:

GraphQL Code Generator and friends

Because this is a schema-level transformation at server start-up time, these fields won’t be available in tools that read your source schema as part of a build step. One popular tool that can read your schema in such a way is the hugely awesome GraphQL code generator when the schema is specified with a *.ts or *.graphl file instead of a URL. In that case, GraphQL codegen provides a way to register plugins that can similarly transform the schema using very similar code save for changes to match typing. Here’s the graphql.config.yml file to configure GraphQL code generator.

schema: src/schema/schema.graphql
extensions:
  codegen:
    generates:
      src/generated/graphql.ts:
        plugins:
          - src/graphql/codegen/plugins/filter-transformer.ts
          - typescript
          - typescript-resolvers

Notice the transformer under the plugins array. This needs to be the first plugin registered so that any changes it makes to the schema are visible to other plugins downstream. Below is the transformer code which is very similar to the one above in logic. I don’t remove the directive or relation type enum here.

type FilterDirectiveArgs = {
  relation?: Relation;
};

const plugin =
  (directiveName: string): PluginFunction =>
  (schema, documents, config) => {
    const rootNode = getCachedDocumentNodeFromSchema(schema);

    visit(rootNode, {
      InputObjectTypeDefinition(node: InputObjectTypeDefinitionNode) {
        const fields = node.fields;
        const fieldsWithDirectives = Object.values(fields).filter(
          (field) =>
            field.directives?.some(
              (directive) => directive.name.value === directiveName,
            ),
        );

        if (fieldsWithDirectives.length === 0) {
          return node;
        }

        const filterFields: string[] = [];

        const relationInputDummyTypes: NamedTypeNode[] = [];

        for (const field of fieldsWithDirectives) {
          const filterDirective = field.directives?.find(
            (directive) => directive.name.value === directiveName,
          );

          if (filterDirective) {
            const args = (
              filterDirective.arguments?.[0]?.value
                ? Object.fromEntries(
                    filterDirective.arguments.map((field) => [
                      field.name.value,
                      (field.value as BooleanValueNode | EnumValueNode).value,
                    ]),
                  )
                : {}
            ) as FilterDirectiveArgs;

            const astType = field.type as NamedTypeNode;

            const isScalar = SupportedScalarTypes.includes(
              astType.name.value as any,
            );

            const isRelation = !!args.relation;

            if (isScalar) {
              filterFields.push(
                buildScalarFilterFields(
                  astType.name.value as any,
                  field.name.value,
                ),
              );
            }

            if (isRelation) {
              let relatedType = field.type as NamedTypeNode;

              filterFields.push(
                buildRelationFilterFields(
                  args.relation,
                  field.name.value,
                  relatedType.name.value,
                ),
              );

              relationInputDummyTypes.push(relatedType);
            }
          }
        }

        const source = `
          scalar Date
          scalar DateTime

          ${relationInputDummyTypes
            .filter((relatedType) => relatedType.name.value !== node.name.value)
            .map((type) => `input ${type.name.value} { id: ID! }`)
            .join("\n")}

          input ${node.name.value} {
              AND: [${node.name.value}!]
              OR: [${node.name.value}!]

              ${filterFields.join("")}
          }
      `;

        const newInputSchema = buildSchema(source);
        const returnType = newInputSchema.getType(
          node.name.value,
        ) as GraphQLInputObjectType;

        (node.fields as any) = returnType.astNode.fields;
      },
    });

    return "";
  };

🔥 with Prisma

Prisma is an open source next-generation ORM. It consists of the following parts:

Prisma Client: Auto-generated and type-safe query builder for Node.js & TypeScript

Prisma Migrate: Migration system

Prisma Studio: GUI to view and edit data in your database.

I like the query syntax of Prisma a lot. You introspect your database using a tool that generates a schema file from which you generate a 💪🏾-ly typed API that you use for CRUD-ing entities to and from your database. I have my issues with Prisma though which I’ll save for another post. The query API shape is a lot like another API I like a lot. It’s been called OpenCRUD. This is a rich query interface packed into a small API surface to allow highly composable specifications of results to be passed kinda like we built above for our API.

When I’ve found 2 independent things I like in the past, I try to create thingamajig out of the 2 of them, and so here’s a simple way to take queries from your GraphQL endpoint and send them straight to Prisma through your GraphQL resolvers.

To simplify this, I implement this for only Scalar fields and no relations and using the flat library. The unflatten function in flat allows keys to be “unflattened” into a nested object. For example,

console.log(
  unflatten({ id_not_in: ["1", "2", "3"] }, { delimiter: "_" })
)
// { id: { not: { in: ["1", "2", "3"] } } }

Using everything from above,

import { unflatten } from "flat";

import { PrismaClient, Prisma } from "@prisma/client";
import { Resolvers } from "../generated/graphql";

const prismaClient = new PrismaClient({
  log: ["query"],
});

const resolvers: Resolvers = {
  Query: {
    books: async (_, { filter }) => {
      const where = unflatten<typeof filter, Prisma.BookWhereInput>(filter, {
        delimiter: "_",
      });

      console.log(where);

      const books = await prismaClient.book.findMany({ where });

      console.log(books);

      return books || [];
    },
  },
};

export default resolvers;

Sample queries and queries sent to Postgres:

{ title: 'Test' }
prisma:query SELECT "public"."Book"."id", "public"."Book"."title", "public"."Book"."author" FROM "public"."Book" WHERE "public"."Book"."title" = $1 OFFSET $2

{ title: { not: { contains: 'Intro' } } }
prisma:query SELECT "public"."Book"."id", "public"."Book"."title", "public"."Book"."author" FROM "public"."Book" WHERE "public"."Book"."title"::text NOT LIKE $1 OFFSET $2

Note: Using the same pattern, you can add a sort directive for generating sort GraphQL enums for your API using a transformer too.

Conclusion

The implementation of GraphQL directives, particularly the custom @filter directive, offers a compelling solution for enhancing the filtering capabilities of GraphQL APIs. The article has explored various filtering options, emphasizing the advantages of a concise, composable, and efficient API through the use of directives.

By introducing a directive-based approach, the article aims to streamline the schema, making it more manageable across teams and reducing verbosity. The proposed solution not only addresses the challenge of complex filtering requirements but also contributes to a more maintainable and readable codebase.

The integration with Prisma showcases a practical application, demonstrating how these directives can seamlessly translate GraphQL queries into database operations. While this approach may not be one-size-fits-all, it provides a valuable tool for developers facing scenarios where compact and composable filtering is a priority.

As with any technical solution, it’s essential to consider the specific needs of your project and weigh the benefits against potential trade-offs. Nevertheless, the exploration of GraphQL directives for supercharging API filtering offers an insightful perspective on optimizing GraphQL schemas for improved efficiency and developer experience.

Hire me: I’m a full-stack engineer with over 10 years of experience in React, Javascript/Typescript, C#, PHP, Python, Rest/GraphQL, XState, SQL (Postgres/MySQL), AWS, and more.
LinkedIn: https://www.linkedin.com/in/abdulkadirna/
Twitter: https://twitter.com/adulkadirna