Data sync source

You can add a custom source to your Optimizely Connect Platform (OCP) app by completing the following steps:

Declare sources in the app manifest file (app.yml).
Declare the schema for your sources (one schema per source).
- src/sources/schema/*.yml or src/sources/{SchemaFunction}.ts
Implement the logic to emit data to sources using the sources.emit() API from any function or job.

Prerequisite

Declare sources in the app manifest file

You can define multiple sources in a single app. You must declare every source in the sources section of the app manifest file (app.yml), and each source must have its own schema.

description – Human-readable description of the source's purpose.
schema – A string referencing a static schema name or an object with entry_point for dynamic schemas.

The following is an example sources section:

sources:
  wordpress_post_sync:
    description: Sync Posts from WordPress
    schema: post
  wordpress_comment_sync:
    description: Sync Comments from WordPress
    schema:
      entry_point: WordpressCommentSchema

📘
Note
Unlike earlier versions of the SDK, you no longer need to define function, lifecycle, or jobs under each source. Instead, use regular functions and jobs with the sources.emit() API to emit data to your sources.

When a user installs your app, any sources you defined in the app displays in the source's Object drop-down list on the Sync Manager page in OCP.

Schema

The schema is where you define the structure of data that flows from your app to the receiving data sync. The source schema is used to configure the data sync field mappings.

Structure

A schema consists of fields and custom types. Fields can use either built-in primitive types (string, integer, boolean, or decimal) or reference custom types that you define.

Schema-level properties:

name – Unique identifier matching the schema value in the manifest.
display_name – User-friendly name.
description – The schema's purpose.
fields – Array of field definitions.
custom_types – (Optional) Array of reusable custom type definitions.

Field properties:

name – Field identifier.
display_name – User-friendly field name.
description – Field's purpose.
type – Field type. Can be primitive, array, or custom type reference.
primary – (Optional) Boolean. Designates the unique identifier (only for primitive types can be primary).

Custom Type properties:

name – Unique identifier for the custom type.
display_name – User-friendly name.
description – What this type represents.
fields – Array of field definitions (same structure as schema fields).

Types

Source schemas support primitive types, arrays, and custom types.

Type Syntax	Description	Example
`string`	Single text value.	`"Product Name"`
`[string]`	Array of text values.	`["tag1", "tag2"]`
`integer`	Single whole number.	`42`
`[integer]`	Array of whole numbers.	`[1, 2, 3]`
`boolean`	True or false value.	`true`
`[boolean]`	Array of booleans.	`[true, false]`
`decimal`	Decimal number.	`19.99`
`[decimal]`	Array of decimals.	`[9.99, 19.99]`
`ExampleCustomType`	Custom type object.	`{id: 1, name: "..."}`
`[ExampleCustomType]`	Array of custom objects.	`[{...}, {...}]`

Type definitions

Source schemas support primitive types, arrays, and custom types:

Primitive Types – Basic data types that include the following:

string – Text data.
boolean – True or false values.
integer – Whole numbers (also aliased as int).
decimal – Floating-point numbers (also aliased as float).

Array Types – Collections using bracket syntax [type].

[string], [integer], [boolean], [decimal] – Arrays of primitive types.
[CustomTypeName] – Arrays of custom type objects.

Custom Types – Reusable structured data types that can be referenced across multiple fields. Useful for complex, nested data structures.

When to Use

Arrays – Use when a field contains multiple values.

Lists of primitive values (tags, IDs, prices).
Collections of structured objects.

Custom types – Use when you have reusable structured data.

Nested objects with multiple fields.
Data structures used in multiple places.
Complex hierarchical models.

Use Cases

Arrays of primitives

Tags/Categories – Use [string] for product tags, article categories, or keywords.
Related IDs – Use [string] or [integer] for related product IDs or category IDs.
Multi-value Attributes – Use [string] for colors, sizes, or other multi-select attributes.
Price History – Use [decimal] for historical pricing data.
Feature Flags – Use [boolean] for per-region or per-variant feature flags.

Custom types

Reusable Address/Contact Information – Define once, use for billing, shipping, or warehouse.
Pricing Structures – Create custom types for pricing tiers, discounts, or promotions.
User/Author Profiles – Define user information that appears in multiple contexts.
Nested Product Variations – Model product variants, options, or configurations.
Hierarchical Categories – Define category structures with multiple levels.

Best practices

Choose the right type.
- Use primitive arrays [string] or [integer] for simple lists.
- Use custom types when objects have multiple related fields.
Use descriptive names – Field and custom type names should clearly indicate what they represent.
Define once, reference everywhere – Avoid duplicating field definitions by using custom types.
Keep types focused – Each custom type should represent a single, cohesive concept.
Document relationships – Use clear descriptions to explain how fields relate to each other.
Consider performance – Large arrays may impact processing.

Declare the schema

The schema is where you define the structure of data that flows through your app, which then displays as options in the data sync field mappings. OCP calls your source class method with properties defined in your schema.

You can choose to implement a static or dynamic schema.

Static – Use when your data structure is fixed and known in advance, all sources of this type share the same data structure, and you want to define the schema declaratively without code.
Dynamic schema – Use when you need to fetch schema information from external systems, and when different instances of the same source might have different fields.

Declare static schema

You must create a matching .yml file in src/sources/schemas for each schema field you defined in the app.yml file.

The following is an example for the post.yml schema:

name: post
display_name: Post
description: External post information
fields:
  - name: external_id
    type: string
    display_name: External ID
    description: Primary identifier in external system
    primary: true
  - name: post_title
    type: string
    display_name: Post Title
    description: The title of the post
  - name: tags
    type: "[string]"
    display_name: Tags
    description: Post tags
  - name: related_post_ids
    type: "[string]"
    display_name: Related Posts
    description: IDs of related posts
  - name: author
    type: Author
    display_name: Author
    description: Post author information
  - name: contributors
    type: "[Author]"
    display_name: Contributors
    description: Additional contributors to the post
custom_types:
  - name: Author
    display_name: Author
    description: Author profile information
    fields:
      - name: author_id
        type: string
        display_name: Author ID
        description: Unique author identifier
      - name: name
        type: string
        display_name: Name
        description: Author's full name
      - name: email
        type: string
        display_name: Email
        description: Author's email address

Declare dynamic schema

You must create and export a class in src/sources/ that extends SourceSchemaFunction. The class name must match the schema entry_point in the app manifest.

Implement the getSourcesSchema() method, which returns a SourceSchema object.

The following shows an example of the app manifest:

sources:
  wordpress_post_sync:
    ...
    schema:
      entry_point: WordpressPostSchema
    ...

The following is an example implementation:

import * as App from '@zaiusinc/app-sdk';

export class WordPressPostSchema extends App.SourceSchemaFunction {
  public async getSourcesSchema(): Promise<App.SourceSchema> {
    try {
      // Fetch a sample post from WordPress API
      const response = await fetch(`${this.config.wordpressUrl}/wp-json/wp/v2/posts?per_page=1`);
      const posts = await response.json();

      // Use the sample post to build our schema
      const samplePost = posts[0];

      // Build schema fields based on sample data
      const fields = [
        // Always include ID as primary field
        {
          name: 'id',
          type: 'string',
          display_name: 'Post ID',
          description: 'WordPress post identifier',
          primary: true,
        },
      ];

      // Add fields based on what is in the sample post
      if (samplePost.title) {
        fields.push({
          name: 'title',
          type: 'string',
          display_name: 'Title',
          description: 'Post title',
        });
      }

      if (samplePost.content) {
        fields.push({
          name: 'content',
          type: 'string',
          display_name: 'Content',
          description: 'Post content',
        });
      }

      if (samplePost.tags && Array.isArray(samplePost.tags)) {
        fields.push({
          name: 'tags',
          type: '[string]',
          display_name: 'Tags',
          description: 'Post tags',
        });
      }

      if (samplePost.author) {
        fields.push({
          name: 'author',
          type: 'Author',
          display_name: 'Author',
          description: 'Post author information',
        });
      }

      if (samplePost.comments && Array.isArray(samplePost.comments)) {
        fields.push({
          name: 'comments',
          type: '[Comment]',
          display_name: 'Comments',
          description: 'Post comments',
        });
      }

      // Define custom types
      const customTypes = [];

      if (samplePost.author) {
        customTypes.push({
          name: 'Author',
          display_name: 'Author',
          description: 'WordPress author information',
          fields: [
            {
              name: 'id',
              type: 'integer',
              display_name: 'Author ID',
              description: 'WordPress author identifier',
            },
            {
              name: 'name',
              type: 'string',
              display_name: 'Name',
              description: 'Author display name',
            },
            {
              name: 'email',
              type: 'string',
              display_name: 'Email',
              description: 'Author email address',
            },
          ],
        });
      }

      if (samplePost.comments && Array.isArray(samplePost.comments)) {
        customTypes.push({
          name: 'Comment',
          display_name: 'Comment',
          description: 'WordPress comment data',
          fields: [
            {
              name: 'id',
              type: 'integer',
              display_name: 'Comment ID',
              description: 'WordPress comment identifier',
            },
            {
              name: 'author_name',
              type: 'string',
              display_name: 'Commenter Name',
              description: 'Name of the commenter',
            },
            {
              name: 'content',
              type: 'string',
              display_name: 'Comment Text',
              description: 'Comment content',
            },
            {
              name: 'approved',
              type: 'boolean',
              display_name: 'Approved',
              description: 'Whether the comment is approved',
            },
          ],
        });
      }

      return {
        name: 'wordpress_post',
        display_name: 'WordPress Post',
        description: 'Blog posts from WordPress with author and comment data',
        fields: fields,
        custom_types: customTypes.length > 0 ? customTypes : undefined,
      };
    } catch (error) {
      // Fallback schema if API request fails
      return {
        name: 'wordpress_post',
        display_name: 'WordPress Post',
        description: 'Blog posts from WordPress',
        fields: [
          {
            name: 'id',
            type: 'string',
            display_name: 'Post ID',
            description: 'WordPress post identifier',
            primary: true,
          },
        ],
      };
    }
  }
}

Implement the logic

Use the sources.emit() API to emit data to your sources from any regular function or job.

Emit data using `sources.emit()`

The sources.emit() function lets you emit data to any source defined in your app manifest from any function or job context.

API signature

import { sources } from '@zaiusinc/app-sdk';

await sources.emit<T>(sourceName: string, data: SourceData<T>): Promise<SourceResponse>;

sourceName – The name of the source as declared in your app.yml file (for example, 'wordpress_post_sync').
data – An object containing the data to emit, matching your source schema.

The data will be fanned out to all active data syncs configured for the app source.

Implement a webhook function (push model)

Use a regular function to receive incoming webhook data and emit it to your source.

App manifest (app.yml)

sources:
  wordpress_post_sync:
    description: Sync Posts from WordPress
    schema: post

functions:
  wordpress_webhook:
    entry_point: WordpressWebhook
    description: Receives WordPress post webhooks

Function implementation (src/functions/WordpressWebhook.ts)

import { Function, Request, Response, sources, logger } from '@zaiusinc/app-sdk';

interface PostData {
  id: string;
  title: string;
  content?: string;
  author?: string;
}

export class WordpressWebhook extends Function {
  public constructor(request: Request) {
    super(request);
  }

  public async perform(): Promise<Response> {
    try {
      const webhookData = this.request.bodyJSON;
      logger.info('Received WordPress webhook', webhookData);

      // Transform data to match your schema
      const postData: PostData = {
        id: webhookData.id,
        title: webhookData.title.rendered,
      };

      // Emit the data to the source
      await sources.emit<PostData>('wordpress_post_sync', { data: postData });

      return new Response(200, { message: 'Success' });
    } catch (error) {
      logger.error('Failed to process webhook', error);
      return new Response(500, { message: error.message });
    }
  }
}

Implement a job (pull model)

Use a regular job to pull data from external systems and emit it to your source.

App manifest (app.yml)

sources:
  wordpress_post_sync:
    description: Sync Posts from WordPress
    schema: post

jobs:
  wordpress_sync:
    entry_point: WordpressSyncJob
    description: Syncs WordPress posts on a schedule
    cron: 0 0 * * * ?

Job implementation (src/jobs/WordpressSyncJob.ts)

import { Job, JobStatus, ValueHash, sources, logger } from '@zaiusinc/app-sdk';

enum JobStep {
  FETCH = 'FETCH',
  DONE = 'DONE',
}

interface SyncJobStatus extends JobStatus {
  state: {
    step: JobStep;
    page: number;
    hasMore: boolean;
  };
}

export class WordpressSyncJob extends Job {
  public async prepare(
    params: ValueHash,
    status?: SyncJobStatus,
    resuming?: boolean
  ): Promise<SyncJobStatus> {
    if (resuming && status) {
      return status;
    }

    return {
      state: {
        step: JobStep.FETCH,
        page: 0,
        hasMore: true,
      },
      complete: false,
    };
  }

  public async perform(status: SyncJobStatus): Promise<SyncJobStatus> {
    switch (status.state.step) {
      case JobStep.FETCH:
        if (!status.state.hasMore) {
          status.state.step = JobStep.DONE;
          break;
        }

        // Fetch a batch of posts from WordPress API
        const response = await this.fetchPostsPage(status.state.page);

        // Emit each post to the source
        for (const post of response.posts) {
          await sources.emit('wordpress_post_sync', { data: post });
        }

        status.state.page += 1;
        status.state.hasMore = response.hasMore;
        break;

      case JobStep.DONE:
        logger.info('WordPress sync complete');
        status.complete = true;
        break;
    }

    return status;
  }

  private async fetchPostsPage(page: number): Promise<{ posts: any[]; hasMore: boolean }> {
    // Implement your API call here
    return { posts: [], hasMore: false };
  }
}

For more details on jobs and it's best practices, see Jobs.

Emit to multiple sources

One of the key benefits of the sources.emit() API is that a single function or job can emit data to multiple sources:

// Emit to different sources based on the data type
await sources.emit('wordpress_post_sync', { data: postData });
await sources.emit('wordpress_comment_sync', { data: commentData });

Best practices

Small work units – Design perform() to handle small units of work that can complete quickly.
Stateful design – Store all necessary state in the state object to enable resumption.
Error handling – Implement proper error handling to ensure jobs fail gracefully and ensure proper try-catch blocks are in place and returns appropriate HTTP status codes.
Interruptibility– Use performInterruptibleTask() and sleep() for operations that may take time.
Progress tracking – Update the state with progress information to show job advancement.

Handle errors

Use appropriate HTTP status codes in your Response objects.
Return detailed error messages to help with debugging.
Implement proper try and catch blocks in your functions.

Complete and publish your app

Legacy approach (deprecated)

⚠️
Deprecated
The following classes are deprecated and will be removed in a future version of the SDK:

SourceFunction

SourceJob

SourceLifecycle

SourceJobStatus, SourceJobInvocation, SourceSleepOptions

Use regular functions and jobs with the sources.emit() API instead. See the Implement the logic section above for the recommended approach.

The legacy approach required defining source-specific entry points in the app manifest and using specialized classes that were tightly coupled to individual data syncs.

Legacy app manifest structure

# DEPRECATED - Do not use for new apps
sources:
  wordpress_post_sync:
    description: Sync Posts from WordPress
    schema: post
    function:
      entry_point: WordpressPostSource
    lifecycle:
      entry_point: WordpressPostLifecycle
    jobs:
      post_sync:
        entry_point: WordpressPostSyncJob
        description: Performs synchronization of posts

Legacy SourceLifecycle class

The SourceLifecycle class was used to manage webhook registration and cleanup for each data sync. This is no longer needed since you can manage webhooks at the app level using the standard Lifecycle class.

// DEPRECATED - Use standard Lifecycle class instead
import * as App from '@zaiusinc/app-sdk';

export class WordpressPostLifecycle extends App.SourceLifecycle {
  public async onSourceCreate(): Promise<App.SourceCreateResponse> {
    // Called when a data sync is created
    return { success: true };
  }

  public async onSourceUpdate(): Promise<App.SourceUpdateResponse> {
    // Called when a data sync is updated
    return { success: true };
  }

  public async onSourceDelete(): Promise<App.SourceDeleteResponse> {
    // Called when a data sync is deleted
    return { success: true };
  }

  public async onSourceEnable(): Promise<App.SourceEnableResponse> {
    // Called when a data sync is enabled
    return { success: true };
  }

  public async onSourcePause(): Promise<App.SourcePauseResponse> {
    // Called when a data sync is paused
    return { success: true };
  }
}

Legacy SourceFunction class

The SourceFunction class was used to handle webhooks in a source-specific context. This is no longer needed since you can use a regular Function with sources.emit().

// DEPRECATED - Use regular Function with sources.emit() instead
import * as App from '@zaiusinc/app-sdk';

export class WordpressPostSource extends App.SourceFunction {
  public async perform(): Promise<App.Response> {
    const webhookData = this.request.body;

    // this.source.emit() was the old way to emit data
    await this.source.emit({ data: webhookData });

    return new App.Response(200, 'Success');
  }
}

Legacy SourceJob class

The SourceJob class was used to run jobs in a source-specific context. This is no longer needed since you can use a regular Job with sources.emit().

// DEPRECATED - Use regular Job with sources.emit() instead
import * as App from '@zaiusinc/app-sdk';

export class WordpressPostSyncJob extends App.SourceJob {
  public async prepare(
    params: App.ValueHash,
    status?: App.SourceJobStatus,
    resuming?: boolean
  ): Promise<App.SourceJobStatus> {
    return { state: {}, complete: false };
  }

  public async perform(status: App.SourceJobStatus): Promise<App.SourceJobStatus> {
    // this.source.emit() was the old way to emit data
    await this.source.emit({ data: { id: '1', title: 'Post' } });

    status.complete = true;
    return status;
  }
}

Migration guide

To migrate from the legacy approach to the new sources.emit() API:

Update your app manifest – Remove function, lifecycle, and jobs from your source definitions. Define regular functions and jobs at the top level instead.

Replace SourceFunction with Function – Create a regular Function class and use sources.emit() to emit data:

// Before (deprecated)
await this.source.emit({ data: myData });

// After (recommended)
await sources.emit('source_name', { data: myData });

Replace SourceJob with Job – Create a regular Job class and use sources.emit():

// Before (deprecated)
await this.source.emit({ data: myData });

// After (recommended)
await sources.emit('source_name', { data: myData });

Remove SourceLifecycle – If you were using lifecycle hooks to register per-data-sync webhooks, consider:
- Registering a single webhook at the app level using the standard Lifecycle class
- Using the app's webhook URL directly without per-data-sync webhook management

Prerequisite

Declare sources in the app manifest file

Note

Schema

Structure

Types

Type definitions

When to Use

Use Cases

Arrays of primitives

Custom types

Best practices

Declare the schema

Declare static schema

Declare dynamic schema

Implement the logic

Emit data using sources.emit()

Implement a webhook function (push model)

Implement a job (pull model)

Emit to multiple sources

Best practices

Handle errors

Complete and publish your app

Legacy approach (deprecated)

Deprecated

Legacy app manifest structure

Legacy SourceLifecycle class

Legacy SourceFunction class

Legacy SourceJob class

Migration guide

Emit data using `sources.emit()`