index.js

/**
 * Data extraction tool for syncing Gadget data to data warehouses via Fivetran.
 * This module provides functionality to extract data from Gadget applications using GraphQL
 * and format it for consumption by data movement platforms like Fivetran.
 */

const {
  compileWithVariableValues,
  Call,
  Var,
} = require("tiny-graphql-query-compiler");

/** Default number of records to fetch per page */
const DefaultPageSize = 100;

/** Models to exclude from data extraction */
const ExcludeModels = ["session"];

/** Field types that should be extracted from Gadget models */
const ExtractFieldTypes = [
  "Any",
  "Array",
  "BelongsTo",
  "Boolean",
  "Code",
  "Color",
  "DateTime",
  "Email",
  "Enum",
  "File",
  "ID",
  "JSON",
  "Money",
  "Null",
  "Number",
  "Object",
  "RecordState",
  "RichText",
  "RoleAssignments",
  "String",
  "URL",
  "Vector",
];

/** GraphQL query to fetch metadata about available models and their fields */
const GadgetMetaQuery = `
  query GadgetMetaQuery {
    gadgetMeta {
      models {
      namespace
        apiIdentifier
        pluralApiIdentifier
        filterGraphQLTypeName
        fields {
          apiIdentifier
          fieldType
        }
      }
    }
  }
`;

/**
 * Executes a GraphQL query against the Gadget API
 * @param {string} query - The GraphQL query to execute
 * @param {Record<string, any>} variables - Variables to be used in the query
 * @returns {Promise<Record<string, any>>} The query result data
 * @throws {Error} If the API request fails or returns errors
 */
async function gadgetGraphQLQuery(query, variables) {
  const url = new URL("/api/graphql", process.env.GADGET_APP_URL);

  const response = await fetch(url, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${process.env.GADGET_API_KEY}`,
    },
    body: JSON.stringify({ query, variables }),
  });

  if (!response.ok) {
    throw new Error(
      `Failed to fetch from Gadget: ${
        response.statusText
      }; ${await response.text()}`
    );
  }

  const result = await response.json();

  if (result.errors || result.data?.errors) {
    throw new Error(
      `Failed to fetch from Gadget: ${[
        ...(result.errors ?? []),
        ...(result.data?.errors ?? []),
      ].join("\n")}`
    );
  }

  return result.data;
}

/**
 * Builds metadata about available Gadget models and their fields
 * @returns {Promise<Array<{
 *   table_name: string,
 *   model_namespace: string[],
 *   model_identifier: string,
 *   model_plural_identifier: string,
 *   filter_graphql_type_name: string,
 *   selection: Record<string, any>
 * }>>} Array of model metadata objects
 */
async function buildGadgetMeta() {
  const result = await gadgetGraphQLQuery(GadgetMetaQuery);

  return result.gadgetMeta.models.flatMap((model) => {
    if (ExcludeModels.includes(model.apiIdentifier)) {
      return [];
    }

    let selection = {};

    for (const field of model.fields) {
      if (ExtractFieldTypes.includes(field.fieldType)) {
        switch (field.fieldType) {
          case "BelongsTo": {
            selection[`${field.apiIdentifier}Id`] = true;
            break;
          }
          case "File": {
            selection[field.apiIdentifier] = {
              url: true,
              fileName: true,
              mimeType: true,
              byteSize: true,
            };
            break;
          }
          case "RichText": {
            selection[field.apiIdentifier] = {
              markdown: true,
            };
            break;
          }
          case "RoleAssignments": {
            selection[field.apiIdentifier] = {
              key: true,
              name: true,
            };
            break;
          }
          default: {
            selection[field.apiIdentifier] = true;
            break;
          }
        }
      }
    }

    return {
      table_name: [...model.namespace, model.pluralApiIdentifier].join("_"),
      model_namespace: model.namespace,
      model_identifier: model.apiIdentifier,
      model_plural_identifier: model.pluralApiIdentifier,
      filter_graphql_type_name: model.filterGraphQLTypeName,
      selection,
    };
  });
}

/**
 * Capitalizes the first letter of a string
 * @param {string} str - The string to capitalize
 * @returns {string} The string with its first letter capitalized
 */
const upperFirst = (str) => str.charAt(0).toUpperCase() + str.slice(1);

/**
 * Queries a Gadget resource with pagination support
 * @param {{
 *   model_namespace: string[],
 *   model_plural_identifier: string,
 *   filter_graphql_type_name: string,
 *   selection: Record<string, any>
 * }} model - Model metadata
 * @param {string} [afterUpdatedAt] - Timestamp to filter records updated after
 * @param {string} [afterCursor] - Cursor for pagination
 * @returns {Promise<{
 *   inserts: Record<string, any>[],
 *   cursor: string,
 *   hasMore: boolean
 * }>} Query results with pagination info
 */
async function queryResource(model, afterUpdatedAt, afterCursor) {
  const pageSize = parseInt(process.env.GADGET_PAGE_SIZE ?? DefaultPageSize);

  const queryVariables = {
    first: Var({ type: "Int!", value: pageSize }),
  };

  if (afterUpdatedAt) {
    queryVariables.filter = Var({
      type: `[${model.filter_graphql_type_name}!]!`,
      value: [{ updatedAt: { greaterThan: afterUpdatedAt } }],
    });
  }

  if (afterCursor) {
    queryVariables.after = Var({ type: "String", value: afterCursor });
  }

  let fields = {
    [model.model_plural_identifier]: Call(queryVariables, {
      pageInfo: {
        hasNextPage: true,
        endCursor: true,
      },
      edges: {
        node: model.selection,
      },
    }),
  };

  for (const namespace of model.model_namespace.reverse()) {
    fields = { [namespace]: fields };
  }

  const { query, variables } = compileWithVariableValues({
    type: `query Get${model.table_name.split("_").map(upperFirst).join("")}`,
    fields,
  });

  const records = await gadgetGraphQLQuery(query, variables);

  return {
    inserts: records[model.model_plural_identifier].edges.map(
      (edge) => edge.node
    ),
    cursor: records[model.model_plural_identifier].pageInfo.endCursor,
    hasMore: records[model.model_plural_identifier].pageInfo.hasNextPage,
  };
}

/**
 * @typedef {Object} InProcessSyncState
 * @property {string} cursor - The cursor value for the next page of records to extract in the current sync
 * @property {boolean} hasMore - Whether there are more records to extract in the current sync
 * @property {string} maxUpdatedAt - The maximum updatedAt value of the last record in the current sync
 */

/**
 * @typedef {Object} FivetranModelState
 * @property {string} [maxUpdatedAt] - The maximum updatedAt value of the last record in the previous sync
 * @property {InProcessSyncState} [inProcessSync] - The state of the current sync for this model
 */

/**
 * @typedef {Record<string, FivetranModelState>} FivetranState - State of each model to sync
 */

/**
 * Fivetran sync handler that manages data extraction and state
 * @param {FivetranState} state - The state of the previous sync
 * @returns {Promise<{
 *   schema: Record<string, any>,
 *   state: FivetranState,
 *   insert: Record<string, any>,
 *   delete: Record<string, any>,
 *   hasMore: boolean
 * }>} Sync results including schema, state, and data
 * @throws {Error} If state is undefined or sync fails
 */
exports.fivetran = async (state) => {
  console.log(JSON.stringify(state, null, 2), "starting fivetran sync");

  if (state === undefined) {
    const error = new Error("No state provided");
    error.status = 400;
    throw error;
  }

  const syncState = {};

  const meta = await buildGadgetMeta();

  const schema = {};
  const inserts = {};
  const deletes = {};
  let hasMore = false;

  for (const model of meta) {
    schema[model.table_name] = { primary_key: ["id"] };
  }

  await Promise.all(
    meta.map(async (model) => {
      const modelState = state[model.table_name] ?? {};

      const newModelState = {
        ...modelState,
      };

      if (!modelState.inProcessSync || modelState.inProcessSync.hasMore) {
        const {
          inserts: modelInserts,
          cursor,
          hasMore: modelHasMore,
        } = await queryResource(
          model,
          modelState.maxUpdatedAt,
          modelState.inProcessSync?.cursor
        );

        if (modelHasMore) {
          hasMore = true;
        }

        inserts[model.table_name] = modelInserts;
        newModelState.inProcessSync = {
          cursor,
          hasMore: modelHasMore,
          maxUpdatedAt: modelInserts.reduce((max, record) => {
            return max && max > record.updatedAt ? max : record.updatedAt;
          }, null),
        };
      }

      syncState[model.table_name] = newModelState;
    })
  );

  let newState = {};

  if (hasMore) {
    newState = syncState;
  } else {
    Object.entries(syncState).forEach(([tableName, modelState]) => {
      newState[tableName] = {
        maxUpdatedAt:
          modelState.inProcessSync?.maxUpdatedAt ?? modelState.maxUpdatedAt,
      };
    });
  }

  return {
    schema,
    state: newState,
    insert: inserts,
    delete: deletes,
    hasMore,
  };
};

/**
 * Google Cloud Function entry point
 * @param {import('express').Request} req - HTTP request context
 * @param {import('express').Response} res - HTTP response context
 */
exports.run = async (req, res) => {
  try {
    const result = await exports.fivetran(req.body.state);
    res.status(200).send(result);
  } catch (error) {
    res.status(error.status ?? 500).send(error.message);
  }
};