Building Content Supply Chain using Workfront, AIO Journaling, AppBuilder Runtime Actions, finally persisting into AEM DAM

Last year we started a content-supply-chain journey to allow non-AEM content providers aka suppliers to supply content into AEM. The suppliers upload assets into dropbox, google drive or own cdn and provide its metadata and asset urls. The content gets syndicated through a 3rd party who was forwarding content updates as json files into AEM. AEM had a workflow that consumed the json, update corresponding content-fragments and ingest assets. This process has been working successfully in Production. I explained in detailed, all the challenges we faced, in my old article: How we migrated a million assets.

This year we are adding one more weapon to our arsenal — Adobe Workfront. Workfront allows suppliers to directly login, view AEM assets and content fragment data, allow them to edit, send for approval, system validation through fusion, and finally ingest the content back into DAM. Since this is Adobe product, Workfront provides seamless integration and avoids the painpoints faced with 3rd party content sync. We used AppBuilder as middleman between Workfront and AEM.

This article will explain step-by-step process for developing App Builder.

Data Dance

Refering architecture diagram above,

• AEM is source of truth persisting Content Fragments and Assets
• Suppliers login into workfront to a ‘Supplier Catalog Report’. They summon for catalog content.
• This summoning goes to Fusion, which invokes AEM persisted graphql query. Fusion caches copy of original graphql response. Text content and images are rendered on workfront report.
• Supplier edits content and posts to Fusion.
• Fusion compares the new record with original graphql response and prepares content payload to events journal.
• This custom event is posted into a journaling queue using https://eventsingress.adobe.io. Workfront job is done.
• At AppBuilder, a non-web Runtime action runs every 5 mins using Whisk Alarm, checking if new events arrived.
• The last event index is stored on aio-lib-state. When new event arrives, the action processes the event payload.
• From Runtime action, it triggers Assets API to create folders, create/update content fragments, create/update/delete assets.
• Next time when supplier summons, AEM graphql returns updated content and closes continues supply chain.

Step 1: Creating App Builder project

AppBuilder is the adobe cloud infrastructure providing powerful serverless platform to build microservices. The microservices are called AIO Runtime Actions. Inorder to get started, we need a AppBuilder project that will deploy and run the code. Following these steps.

• Login to developer.adobe.com. Click on create new project

• This will create skeleton project. Download the Project json, will be used later.
• Next should create Bootstrap App using CLI. Run command aio app init wf-payload-processor . Open the new project from VSC. Project structure looks like this (ignore some additional files. will explain later)

• Locate the index.js and add simple response message to test, like this

function main(params) {
  return {
    headers: { "Content-Type": "application/json" },
    statusCode: 200,
    body: { result: "Runtime action executed successfully" },
  };
}

const _main = main;
export { _main as main };

• Deploy this action to AppBuilder using command aio app deploy . Windows support a--local param. I haven’t tried though.

• Upon successful deployment, a webaction url is generated. To test, copy the webaction url and execute POST using postman

Our Runtime action is ready. Lets add more code and configuration.

Why AIO Events Journaling API?

Why not custom API: The edits made by Supplier at Workfront needs to be persisted into AEM. This can be very easily done by exposing an AEM servlet and workfront can call servlet with payload. But definite NO. I explained the challenges faced with synchronous API calls into AEM in my previous article under Attempt 3.

In nutshell, REST APIs have limitations, have to implement retries, run into rate limits, long running API calls will hang / crash AEM server, and most important, introduces coupling. Any changes required to API needs AEM deployment, microservice design helps to decouple application dependencies.

Verdict: We wanted an event-driven pub-sub model instead of monolith API-driven.

Why not WF-AEM Connector: Another consideration to ingrate workfront with AEM is to use ootb connector. We did setup the connector,

This performs 2-way, shows assets from DAM on WF and viceversa WF can upload assets directly into DAM folders.

But this connector did not help our use case, since we needed a Fusion layer to perform validations, wait for approvals, and push only valid content into AEM. Didnt want suppliers to directly upload assets into DAM, instead after approvals and validation, wanted application to upload.

Why not other eventing solutions: Having decided, we want to pursue event-driven, Adobe still recommends 4 techniques

After evaluation, we settled with Journaling API. Here was the reasons

Important consideration was - reprocessing ability. In past, we had production incidents with deployment issues, code bugs, supplier human errors etc. This mandated requirement for, a backdoor to reprocess events. Journaling API persists events upto 7 days. This option suited us.

Step 2: Setup Journaling

Adobe documentation explains the steps. We can setup journaling using

1. CLI
2. SDK
3. API

I ll show the third way. Run the below curl commands in same order

Note: Make sure to store the response of each request below. Will be needed for step 5.

1. Generate token

curl --location 'https://ims-na1.adobelogin.com/ims/token/v3' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'grant_type=client_credentials' \
--data-urlencode 'client_id=<id>' \
--data-urlencode 'client_secret=<secret>' \
--data-urlencode 'scope=AdobeID,openid,read_organizations,additional_info.projectedProductContext,additional_info.roles,adobeio_api,read_client_secret,manage_client_secrets'

The values for generating token to be copied from AppBuilder runtime Oauth token

2. Create Custom Events Provider. Use provider API.

curl -i -v --request POST \
  --url https://api.adobe.io/events/${consumerId}/${projectId}/${workspaceId}/providers \
  --header "x-api-key: $api_key" \
  --header "Authorization: Bearer $oauth_s2s_token" \
  --header 'content-type: application/json' \
  --header 'Accept: application/hal+json' \
  --data '{
      "label": "a label of your choice for you Custom Events Provider",
      "description": "a description of your Custom Events Provider",
      "docs_url": "https://yourdocumentation.url.if.any"
    }'

For the docs_url, I gave link to our internal confluence page to project. So on the developer console, the event provider links to our internal confluence page

Adobe documentation provides help where to find the various parameters needed to setup custom provider.

3. Create Provider Metadata. — This step creates new custom event_code.

curl -i -v --request POST \
  --url  https://api.adobe.io/events/${consumerId}/${projectId}/${workspaceId}/providers/${providerId}/eventmetadata \
  --header "x-api-key: $api_key" \
  --header "Authorization: Bearer $oauth_s2s_token" \
  --header 'content-type: application/json' \
  --header 'Accept: application/hal+json' \
    --data '{
      "event_code": "aem.custom.supplier.content.dev",
      "label": "Event type used to upsert supplier content into AEM Dev environment",
      "description": "This event gets published from AEM and Workfront, event contains payload of supplier data and upserts into AEM"
    }'

4. Finally register event_code — This step will bind the provider created in step1 with event created in step3. So the events pushed into journal, will be subscribed by this provider. Can view those events from Events Browser

Journaling is ready. Use Events API for reference. Lets test if this setup is working correctly. To test

5. Post a test event — Use the events ingres API to post event.

curl -i --location --request POST  \
       --url https://eventsingress.adobe.io \
       --header "x-api-key: $api_key" \
       --header "Authorization: Bearer $oauth_s2s_token" \
       --header 'Content-Type: application/cloudevents+json' \
       --header "x-event-phidata: $is_phidata" \
       --data '{
         "datacontenttype": "application/json",
         "specversion": "1.0",
         "source": "urn:uuid:'"${provider_id}"'",
         "type": "'"${event_code}"'",
         "id": "'"${event_id}"'",
         "data": "your event json payload"
       }'

The values needed to post, should be picked from previous 4 responses. After successfully pushing an event into journal, validate the event by fetching from journaling using this curl.

curl --location 'https://api.adobe.io/events/organizations/orgid/integrations/credid/registrationid?fetchResponseHeaders=true' \
--header 'x-ims-org-id: orgid' \
--header 'x-api-key: clientid' \
--header 'Authorization: Bearer token'

Our configuration to setup journaling is done. We tested by posting an event into journal and validated by fetching back. Next we ll create Runtime action to process the events.

Step 3: Create Runtime Action

From step1 we created only skeleton action. We ll further extend the action to listen the journaling queue and process the event payload.

We are renaming all .js into .mjs for ESM Module support. To make it work, we point to index.mjs at app.config.yaml as function: actions/generic/index.mjs

index.mjs — The index.mjs initializes global params. I am reading the inputs from app.config.yaml and setting global variables so as to use at different modules. The index.mjs just invokes startFetchingEvents function at journalevents.mjs

import { Core } from "@adobe/aio-sdk";
import { startFetchingEvents } from "./journalevents.mjs";
import { setGlobals } from "./common.mjs";

function main(params) {
  const logger = Core.Logger("main", { level: params.LOG_LEVEL || "info" });
  setGlobals(params);
  try {
    startFetchingEvents();
  } catch (error) {
    logger.error(`error happened ${JSON.stringify(error)}`);
  }
  return {
    headers: { "Content-Type": "application/json" },
    statusCode: 200,
    body: { result: "Runtime action executed successfully" },
  };
}

const _main = main;
export { _main as main };

journalevents.mjs — This module reads events from journal using SDK and invokes payload processors aka assets.mjs, folder.mjs and contentfragment.mjs to perferm respective crud operations

import { Core } from "@adobe/aio-sdk";
import stateLib from "@adobe/aio-lib-state";
import sdk from "@adobe/aio-lib-events";
import { filter, concatMap } from "rxjs/operators";
import { of } from "rxjs";
import { getTokenUsingSDK } from "./token.mjs";

import {
  STATE_KEY_JOURNAL_EVENT_INDEX,
  SEVEN_DAYS_IN_SECONDS,
  clientId,
  consumerOrgId,
  credId,
  imsOrgId,
  oauthToken,
  oauthTokenExpiry,
  registrationId,
  setOauthToken,
} from "./common.mjs";
import { uploadAsset, deleteAsset } from "./assets.mjs";
import { createFolder } from "./folder.mjs";
import { updateContentFragmentsUsingAssetsAPI } from "./contentfragment.mjs";

const logger = Core.Logger("main");

async function saveEventIndexToState(state, evt) {
  if (evt && evt.position) {
    await state.put(STATE_KEY_JOURNAL_EVENT_INDEX, evt.position, {
      ttl: SEVEN_DAYS_IN_SECONDS,
    });
    logger.log("Saved index: " + evt.position);
  }
}
async function processEvent(evt) {
  let payload = {};
  logger.log(
    `Processing new events: ${JSON.stringify(evt)} of datatype ${typeof evt
      .event.data}`
  );
  try {
    if (evt.event.data && typeof evt.event.data === "string") {
      // It's safe to use replace single quotes to double quotes 
      const payloadStr = evt.event.data?.replace(/'/g, '"');
      payload = JSON.parse(payloadStr);
    } else if (evt.event.data && typeof evt.event.data === "object") {
      payload = evt.event.data;
    } else {
      // Handle the case where evt.event.data is not a string
      logger.error(
        `Event data is not a string or json object on: ${JSON.stringify(evt)}`
      );
      return;
    }
  } catch (e) {
    logger.error(`Failed to parse payload: ${e.message}`);
    return;
  }
  if (payload.createFolder) {
    await createFolder(payload.createFolder);
  }
  if (payload.createAsset) {
    uploadAsset(payload.createAsset);
  }
  if (payload.updateAsset) {
    uploadAsset(payload.updateAsset);
  }
  if (payload.deleteAsset) {
    deleteAsset(payload.deleteAsset);
  }
  if (payload.updateCF) {
    updateContentFragmentsUsingAssetsAPI(payload.updateCF);
  }
}
async function fetchEventsUsingSDK(state, since = "") {
  logger.info("Checking events after index: " + since);
  const client = await sdk.init(imsOrgId, clientId, oauthToken);
  const baseURL = `https://api.adobe.io/events/organizations/${consumerOrgId}/integrations/${credId}/${registrationId}`;
  const journalOptions = since ? { since: since } : undefined;
  const journalObservable = await client.getEventsObservableFromJournal(
    baseURL,
    journalOptions
  );
  journalObservable
    .pipe(
      filter((evt) => evt.position && evt.event.data),
      concatMap((evt) => {
        return of(evt).pipe(
          concatMap((event) => processEvent(event).then(() => event))
        );
      })
    )
    .subscribe(
      async (x) => await saveEventIndexToState(state, x), // any action onNext event
      (e) => logger.error("onError: " + e.message), // any action onError
      () => logger.log("onCompleted") //action onComplete
    );
}

export async function startFetchingEvents() {
  try {
    if (!oauthTokenExpiry || Date.now() >= oauthTokenExpiry) {
      const { access_token, expires_in } = await getTokenUsingSDK();
      setOauthToken(access_token, expires_in);
    }
    const state = await stateLib.init();
    const index = await state.get(STATE_KEY_JOURNAL_EVENT_INDEX);
    const indexValue = index?.value || "";
    await fetchEventsUsingSDK(state, indexValue);
  } catch (error) {
    logger.error("Error obtaining token:" + error.message);
  }
}

Here getTokenUsingSDK is used to fetch Oauth token. The token is set into a global variable to be reused, incase events occur within the runtime action lifetime.

Next we need to persist the last event index in DB/storage outside the Runtime action. So when new action spins up, the last event index is still available at persistence. For this purpose, we use State library as explained in this Code Lab example.

Next function is fetchEventsUsingSDK . This function is using aio-lib-events SDK to fetch events. Now events can be fetched in 2 ways.

1. API. By passing ?since param into GET request https://api.adobe.io/events/organizations/org/integrations/consumerid/regid?since=&limit=1, we can read next event. But when there is burst of events, the function needs to be called recursively until all events are processed. This is specifically useful if events needs to be processed in order.
2. SDK. — This is the recommended approach using the RxJS observables. Now one issue I found with SDK was, the events were NOT sequential. By default events are consumed asynchronously. This was issue to me, since I wanted create events BEFORE update/delete events. So I figured a way to make the SDK consume events synchronously instead of asynchronous. The issue and solution is explained in community here.

Each Events should Await for completion before proceeding to next event:

Even though event-driven design educates to process events in parallel and asynchronously, one problem I faced was, when multiple events were attempting to update same Content Fragment or asset at AEM, it resulted in 409 conflict. The RxJS observable clears the journal and consumes all events and stacks them asynchronously. Though nodejs is single-threaded, the async cannot avoid 409 conflicts. I wanted reliable way to complete processing an event before moving to next event.

SDK code to process asynchronous:

const sdk = require('@adobe/aio-lib-events')

async function sdkTest() {
  // initialize sdk
  const client = await sdk.init('<organization id>', 'x-api-key', '<valid auth token>', '<http options>')
  // get the journalling observable
  const journalling = client.getEventsObservableFromJournal('<journal url>', '<journalling options>')
  // call methods
  const subscription = journalling.subscribe({
    next: (v) => console.log(v), // Action to be taken on event
    error: (e) => console.log(e), // Action to be taken on error
    complete: () => console.log('Complete') // Action to be taken on complete
  })

  // To stop receiving events from this subscription based on a timeout
  setTimeout(() => subscription.unsubscribe(), <timeout in ms>)
}

This is the example snippet. The next callback will be triggered asynchronous.

Modified code:

const journalObservable = await client.getEventsObservableFromJournal(
    baseURL,
    journalOptions
  );
  journalObservable
    .pipe(
      filter((evt) => evt.position && evt.event.data),
      concatMap((evt) => {
        return of(evt).pipe(
          concatMap((event) => processEvent(event).then(() => event))
        );
      })
    )
    .subscribe(
      async (x) => await saveEventIndexToState(state, x), // any action onNext event
      (e) => logger.error("onError: " + e.message), // any action onError
      () => logger.log("onCompleted") //action onComplete
    );

Here I introduced .pipe function to first consume the event, complete its processing and finally onNext callback is used to store the successfully processed event index into aio-lib-state. One point with this approach is, incase event processing failing (if AEM down or payload is huge to process with action lifetime), the event index fails to persist. So next time the action reruns same event index. This helps me as free retry, wherein action executes every 5mins. If event failed to process, next action will retry same payload. Flip side, if payload was bad, it clogs the queue. I made another arrangement to unclog. Wrote another action to reset event index. So our maintenance team will reset event index when blocking payload causes traffic jam.

Process Event Function:

The event payload is a structured schema provided to Fusion team. The payload must include one of these objects

if (payload.createFolder) {
    await createFolder(payload.createFolder);
  }
  if (payload.createAsset) {
    uploadAsset(payload.createAsset);
  }
  if (payload.updateAsset) {
    uploadAsset(payload.updateAsset);
  }
  if (payload.deleteAsset) {
    deleteAsset(payload.deleteAsset);
  }
  if (payload.updateCF) {
    updateContentFragmentsUsingAssetsAPI(payload.updateCF);
  }

And its corresponding operation is fired. Note createFolder function alone is awaited to make sure folders are created before other create/update/deletes

The rest code is straight, uses Assets API and AssetCompute microservice for crud operation into AEM.

Assets is a VERY expensive module

This is my assets module. Looks straight implementation using AssetCompute microservice.

assets.mjs

import {
  ASSETS_API,
  base64Credentials,
  INITIATE_UPLOAD_URL,
  PARENT_PATH,
  TARGET_SERVER,
} from "./common.mjs";
import { Core } from "@adobe/aio-sdk";
import { checkIfFolderExists, getParentFolder } from "./folder.mjs";

const logger = Core.Logger("main");

const makeCompleteUploadRequest = async (
  uploadToken,
  mimeType,
  fileName,
  completeUri
) => {
  try {
    const formData = new URLSearchParams();
    formData.append("uploadToken", uploadToken);
    formData.append("fileName", fileName);
    formData.append("createVersion", "true");
    formData.append("versionLabel", fileName);
    formData.append("mimeType", mimeType);
    formData.append("description", "dc:description value");
    formData.append("title", "dc:title value");

    const url = `${TARGET_SERVER}${completeUri}`;
    const response = await fetch(url, {
      method: 'POST',
      headers: {
        "Content-Type": "application/x-www-form-urlencoded",
        Authorization: `Basic ${base64Credentials}`,
      },
      body: formData.toString()
    });

    if (!response.ok) {
      throw new Error(`HTTP error! status: ${response.status}`);
    }

    const responseData = await response.json();
    logger.log(`Upload completed successfully for ${fileName}`);
    return responseData;
  } catch (error) {
    logger.error(`Error completing upload for ${fileName}: ${error.message}`);
    throw error;
  }
};

async function initiateUpload(asset) {
  try {
    const { itemnumber, fileName, sourceUrl, assetFolder } = asset;

    const itemFolder = getParentFolder(itemnumber);
    const parentFolder = `${PARENT_PATH}/${itemFolder}/${assetFolder}`;
    const folderExists = await checkIfFolderExists(
      `${TARGET_SERVER}${ASSETS_API}${itemFolder}/${assetFolder}`
    );
    if (!folderExists) {
      logger.error(
        `Folder ${parentFolder} does not exist. Unable to import ${fileName} for item ${itemnumber}`
      );
      return;
    }
    logger.log(`downloading ${fileName} for folder ${parentFolder}`);
    const assetBinaryResponse = await fetch(sourceUrl);
    if (!assetBinaryResponse.ok) {
      throw new Error(`Failed to download asset: ${assetBinaryResponse.statusText}`);
    }
    const assetBinary = await assetBinaryResponse.arrayBuffer();
    const fileSize = assetBinaryResponse.headers.get("content-length");
    logger.log(`downloaded ${fileName}. Now uploading to ${parentFolder}`);

    const initiateUploadConfig = {
      method: "POST",
      headers: {
        Authorization: `Basic ${base64Credentials}`,
        'Content-Type': 'application/x-www-form-urlencoded'
      },
      body: new URLSearchParams({
        fileName: fileName,
        fileSize: fileSize
      })
    };

    const initiateUploadResponse = await fetch(
      `${TARGET_SERVER}${parentFolder}${INITIATE_UPLOAD_URL}`,
      initiateUploadConfig
    );

    if (!initiateUploadResponse.ok) {
      throw new Error(`Failed to initiate upload: ${initiateUploadResponse.statusText}`);
    }

    const initiateRes = await initiateUploadResponse.json();
    const { files, completeURI } = initiateRes;
    if (!files || files.length === 0) {
      throw new Error("No file information found in the response");
    }
    const { uploadToken, fileName: fileNameForCompleteRequest, mimeType, minPartSize, uploadURIs } = files[0];
    if (!uploadURIs || uploadURIs.length === 0) {
      throw new Error("No upload URIs found in the response for the file");
    }

    if (uploadURIs.length === 1) {
      const putResponse = await fetch(uploadURIs[0], {
        method: 'PUT',
        headers: {
          "Content-Type": mimeType,
        },
        body: assetBinary
      });

      if (!putResponse.ok) {
        throw new Error(`Put Binary failed: ${putResponse.statusText}`);
      }
    } else {
      for (let index = 0; index < uploadURIs.length; index++) {
        const start = index * minPartSize;
        let end = start + minPartSize;
        if (end > fileSize) {
          end = fileSize;
        }
        const partBlob = assetBinary.slice(start, end);
        const putResponse = await fetch(uploadURIs[index], {
          method: 'PUT',
          headers: {
            "Content-Type": mimeType,
          },
          body: partBlob
        });
        if (!putResponse.ok) {
          throw new Error(`Put Binary failed for part ${index + 1}: ${putResponse.statusText}`);
        }
      }
    }

    await makeCompleteUploadRequest(
      uploadToken,
      mimeType,
      fileNameForCompleteRequest,
      completeURI
    );
    logger.log(`${fileName} uploaded successfully under ${parentFolder}`);
  } catch (err) {
    logger.error(`Error in uploading asset: ${err.message}`);
    throw err;
  }
}

async function initiateDelete(asset) {
  try {
    const { itemnumber, fileName, assetFolder } = asset;
    const itemFolder = getParentFolder(itemnumber);
    const assetPath = `${itemFolder}/${assetFolder}/${fileName}`;
    const url = `${TARGET_SERVER}${ASSETS_API}${assetPath}`;

    logger.log(`Deleting asset: ${fileName} from ${assetPath}`);

    const deleteResponse = await fetch(url, {
      method: "DELETE",
      headers: {
        Authorization: `Basic ${base64Credentials}`,
      },
    });

    if (deleteResponse.status === 204) {
      logger.log(`Asset ${fileName} deleted successfully from ${assetPath}`);
    } else {
      logger.error(
        `Failed to delete asset ${fileName}. Status: ${deleteResponse.status}`
      );
    }
  } catch (err) {
    logger.error(`Error in deleting asset: ${JSON.stringify(err)}`);
  }
}

export const uploadAsset = (assets) => {
  assets.forEach((asset) => {
    initiateUpload(asset);
  });
};

export const deleteAsset = (assets) => {
  assets.forEach((asset) => {
    initiateDelete(asset);
  });
};

The main issue lies in the line

const assetBinaryResponse = await fetch(sourceUrl);

Before we upload using AssetCompute microservice, the second step of PUT operation, expects full/part of binary. In order to ship the binary into AEM DAM, the runtime action needs to download entire payload, buffer cache, possible chuck the payload and ship into AEM. This is an expensive operation, especially when suppliers upload huge assets or videos.

Mitigation 1: The Step1 initiateUpload provides an array of PUT urls. So immediate mitigation was to chunk the binary and place multiple PUT requests into AEM. But this doesnt solve when asset is large say >200MB

Mitigation 2: Problem with download and buffering large assets is, AIO has default system settings.

My action is non-blocking since its invoked by Whisk Alarm, and not browser / API. But all actions has memory limit of 250MB. So if we attempt to download large video assets, action will run out of memory and crash. To mitigate this, we can change default settings at app.config.yaml like this

application:
  actions: actions
  runtimeManifest:
    packages:
      asset-uploader:
        license: Apache-2.0
        actions:
          generic:
            function: actions/generic/index.mjs
            runtime: nodejs:18
            inputs:
              LOG_LEVEL: debug
              MY_CLIENTID: $MY_CLIENTID
              MY_CLIENTSECRET: $MY_CLIENTSECRET
              MY_IMSORGID: $MY_IMSORGID
              MY_CONSUMERORGID: $MY_CONSUMERORGID
              MY_CREDENTIALID: $MY_CREDENTIALID
              MY_REGISTRATIONID: $MY_REGISTRATIONID
              MY_TECHNICALACCOUNTID: $MY_TECHNICALACCOUNTID
              MY_TECHNICALACCOUNTEMAIL: $MY_TECHNICALACCOUNTEMAIL
              MY_TARGET_SERVER: $MY_TARGET_SERVER
            annotations:
              require-adobe-auth: false
              final: true
            limits:
             timeout: 1200000
             memory: 4096
        triggers:
          fivemininterval:
            feed: /whisk.system/alarms/interval
            inputs: 
              minutes: 5
        rules:
          fiveMinuteRule:
            trigger: fivemininterval
            action: generic

Here

       limits:
             timeout: 1200000
             memory: 4096

This config helps to change the default settings. This way we can run longer and process larger payloads upto 4GB. But still this is not adequate, since such large event becomes blocking payload. Causes traffic jam by blocking event traffic.

Mitigation 3: Working with Adobe CSM, we received the best recommendation from Adobe, a dedicated assets ingestor microservice. This is an experimental API. This API wraps the AssetCompute microservice, but runs on dedicated adobe infrastructure. The request looks like this

https://author-program-environment.adobeaemcloud.com/adobe/assetimport/from-url

Purposefully url is obscured, since its still experimental. But this API accepts a fromUrl and target folder, imports the asset independently and uploads safely into AEM. Hence the action makes a send — n — forget call to this new API that ll guarantee upload asset into DAM.

Content Fragment Module

The CF module also had interesting findings. Adobe provides 2 ways to update CF: AssetsAPI and OpenAPI. I had to choose Creates using OpenAPI and Updates using AssetsAPI. WHY??

contentfragment.mjs

import { Core } from "@adobe/aio-sdk";
import {
  ASSETS_API,
  BASIC_AUTH_HEADER,
  CF_API,
  PARENT_PATH,
  CF_MODAL,
  TARGET_SERVER,
} from "./common.mjs";
import { getParentFolder } from "./folder.mjs";
const logger = Core.Logger("main");
export async function createContentFragmentUsingOpenAPI(itemnumber) {
  const parentFolder = getParentFolder(prop);

  logger.log(
    `Creating CF for ${itemnumber} in ${TARGET_SERVER}${ASSETS_API}${parentFolder}`
  );
  const fields = [
    {
      name: "customprop",
      values: [prop],
      type: "text",
    },
  ];
  const raw = JSON.stringify({
    title: prop,
    modelId: CF_MODAL,
    parentPath: `${PARENT_PATH}/${parentFolder}`,
    fields: fields,
  });

  const url = TARGET_SERVER + CF_API;
  try {
    const cfResponse = await fetch(url, {
      method: "POST",
      headers: BASIC_AUTH_HEADER,
      body: raw,
    });

    if (!cfResponse.ok) {
      throw new Error(`HTTP error! status: ${cfResponse.status}`);
    }

    await cfResponse.json();
  } catch (e) {
    logger.error(
      `error creating CF with reason: ${e.message} and ${JSON.stringify(e)}`
    );
  }
}

export const updateContentFragmentsUsingAssetsAPI = (items) => {
  items.forEach(({ itemnumber, elements }) =>
    updateContentFragmentUsingAssetsAPI(itemnumber, elements)
  );
};

export async function updateContentFragmentUsingAssetsAPI(
  itemnumber,
  elementList
) {
  logger.log(`Updating CF for ${itemnumber}`);
  const parentFolder = getParentFolder(itemnumber);
  const transformedElementList = Object.entries(elementList).reduce(
    (property, [key, value]) => {
      property[`salsify:${key}`] = {
        value: value,
        ":type": "string", // All Properties are stored as simple string values
      };
      return property;
    },
    {}
  );
  const elements = {
    properties: {
      elements: transformedElementList,
    },
  };
  try {
    const resp = await fetch(
      `${TARGET_SERVER}${ASSETS_API}${parentFolder}/${itemnumber}`,
      {
        method: "PUT",
        headers: {
          ...BASIC_AUTH_HEADER,
          "Content-Type": "application/json",
        },
        body: JSON.stringify(elements),
      }
    );

    if (!resp.ok) {
      throw new Error(`HTTP error! status: ${resp.status}`);
    }

    await resp.json();
  } catch (e) {
    logger.error(
      `error updating CF of itemnumber ${itemnumber}: ${JSON.stringify(e)}`
    );
  }
}

Now the issue I faced, I converted into an Idea and posted in community. The problem calling OpenAPI to perform updates was: the OpenAPI mandates a If-Match header check. To cater this, we must place 3 trips

1. Find CF ID using List
2. Find etag using Get
3. Edit CF using Patch

Somehow the Adobe API documentation was NOT matching its implementation. Not sure why, but documentation reads List must return etag, but when tested from postman, it wasn’t returning. Raised Adobe ticket and couldn’t find reason. So temporary solution, using AssetsAPI for updates. Once this adobe bug is resolved, will migrate to OpenAPI.

Conclusion

Overall, our goal was to establish an event-driven design to allow Workfront and future 3rd parties to ingest content into AEM. AppBuilder provides infrastructure to build microservices that listens to update events and ingests content into AEM. This helps to construct the Content-Supply-Chain from workfront into AEM and later to downstream delivery systems.

I ll keep adding more articles with my findings through this journey.

Happy Coding!