Update Local Subscriptions with GraphQL Response Upon Outbox Mutations Processed

[martin-minovski]

Description

I've encountered an issue with the DataStore plugin when working with a GraphQL API. I have set up an AWS Amplify app, and for two of my models, I've added a custom resolver function to the create mutation pipeline resolver just before the final resolver function that saves data to DynamoDB. This custom resolver function modifies some of the model attributes.

The problem I'm observing is that while the GraphQL create mutation response correctly reflects the modified data, the DataStore subscriptions are not updating the Flutter widgets to match this response. Instead, they persist the unmodified data from the saved model. To force the UI to correctly display the data that's stored on the cloud, I have to restart the app.

Currently, I have a workaround which looks like this:

StreamSubscription<DataStoreHubEvent>? _hubSubscription;

void observeEvents() {
  _hubSubscription = Amplify.Hub.listen(HubChannel.DataStore, (hubEvent) async {
    if (hubEvent.eventName == 'outboxMutationProcessed') {
      final status = hubEvent.payload as OutboxMutationEvent?;
      if (status!.modelName == 'ChatMessage') {
        await stopObservingMessages();
        observeMessages();
      }
      if (status!.modelName == 'ChatThread') {
        await stopObservingThreads();
        observeThreads();
      }
    }
  });
}

void observeMessages() {
  _messageStream = Amplify.DataStore.observeQuery(
    ChatMessage.classType,
    sortBy: [ChatMessage.SENTAT.descending()],
  ).listen((QuerySnapshot<ChatMessage> snapshot) {
    _chatMessages = snapshot.items;
    notifyListeners();
  });
}

void observeThreads() {
  _threadStream = Amplify.DataStore.observeQuery(
      ChatThread.classType
  ).listen((QuerySnapshot<ChatThread> snapshot) {
    _chatThreads = snapshot.items;
    notifyListeners();
  });
}

Future<void> stopObservingMessages() async {
  if (_messageStream == null) return;
  await _messageStream?.cancel();
  _messageStream = null;
}

Future<void> stopObservingThreads() async {
  if (_threadStream == null) return;
  await _threadStream?.cancel();
  _threadStream = null;
}

However, this solution is far from ideal, as it triggers a scan on all items with every sent message.

Wouldn't it be more efficient and intuitive if the local subscriptions were updated with the GraphQL response when the 'outboxMutationProcessed' event occurs? This way, local views would reflect the true data state as modified by the backend resolvers.

Categories

• [ ] Analytics
• [ ] API (REST)
• [X] API (GraphQL)
• [ ] Auth
• [ ] Authenticator
• [X] DataStore
• [ ] Notifications (Push)
• [ ] Storage

Steps to Reproduce

No response

Screenshots

No response

Platforms

• [X] iOS
• [X] Android
• [ ] Web
• [ ] macOS
• [ ] Windows
• [ ] Linux

Flutter Version

3.10.1

Amplify Flutter Version

1.4.0

Deployment Method

Amplify CLI

Schema

No response

[Jordan-Nelson]

Hi @martin-minovski - Would you be able to share your custom resolver and schema definition?

[martin-minovski]

Hi @martin-minovski - Would you be able to share your custome resolver and schema definition?

Sure!

Schema:

type ChatMessage @model @auth(rules: [
  {allow: owner, operations: [create, read]},
  {allow: private, provider: iam, operations: [create, read, update]}
]) {
  id: ID!
  message: String!
  seen: Boolean!
  sentAt: AWSDateTime!
  threadID: String
  owner: String @auth(rules: [
    {allow: owner, operations: [read]},
    {allow: private, provider: iam, operations: [create, read]}
  ])
}

type ChatThread @model @auth(rules: [
  {allow: owner, operations: [create, read]},
  {allow: private, provider: iam, operations: [create, read, update]}
]) {
  id: ID!
  subject: String!
  from: String
  slackThreadId: String
  openedAt: AWSDateTime!
  closedAt: AWSDateTime
  priority: Priority!
  firstMessage: String!
  owner: String @auth(rules: [
    {allow: owner, operations: [read]},
    {allow: private, provider: iam, operations: [create, read]}
  ])
}

enum Priority {
  HIGH
  MEDIUM
  LOW
}

Chat message custom resolver:

module.exports.handler = async (event, context) => {
    let message = event.input;
    let stash = event.stash;

    let createdAt = stash.defaultValues.createdAt;
    message.sentAt = createdAt;

    let response = await getSlackThreadIdByThreadId(message.threadID);
    let slackThreadId = response.data.getChatThread.slackThreadId;

    // Submit message to Slack
    await slack_addMessageToThread(slackThreadId, message.message);

    return message;
};

Chat thread custom resolver:

module.exports.handler = async (event, context) => {
    let thread = event.input;
    let stash = event.stash;

    let from = thread.from;
    let createdAt = stash.defaultValues.createdAt;
    let subject = thread.subject;

    thread.openedAt = createdAt;

    // Submit thread to Slack and get the slack thread ID
    thread.slackThreadId = await slack_createChatThread(from, subject, test); 

    return thread;
};

Custom resolver request mapping template:

#set($args = $util.defaultIfNull($ctx.stash.transformedArgs, $ctx.args))
{
  "version" : "2017-02-28",
  "operation": "Invoke",
  "payload": {
    "input": $util.toJson($args.input),
    "stash": $util.toJson($ctx.stash)
  }
}

Custom resolver response mapping template:

#if($ctx.error)
  $util.error($ctx.error.message, $ctx.error.type)
#else
  #set($ctx.stash.tweakedInput = $ctx.result)
  $util.toJson($ctx.result)
#end

Modifications made to final DynamoDB resolver:

• Mutation.createChatThread.req.vtl: replace $args.input with $ctx.stash.tweakedInput
• Mutation.createChatMessage.req.vtl: replace $args.input with $ctx.stash.tweakedInput

[Jordan-Nelson]

Hi @martin-minovski - Apologies for the delay. This is related to https://github.com/aws-amplify/amplify-flutter/issues/1355.

When a new item is created on the same device as the observeQuery subscription, observeQuery emits an event immediately with the data from the local create operation. There is no new event emitted once the item has been saved to app sync. Typically, the events are identical with the exception of createdAt/updatedAt metadata. But when a custom resolver mutates the item, this is not the case.

It seems reasonable in this scenario that observeQuery would emit a second event after the successful save operation with the updated info. Would that satisfy your use case?

[Jordan-Nelson]

After discussing this internally, I am going to mark this as a bug. The correct behavior should be to emit a new snapshot in this case.

[Jordan-Nelson]

@martin-minovski - This should be resolved in the latest version of amplify flutter. If you are still facing issue with the latest version please let us know. Otherwise we will close this out.

[haverchuck]

Closing this issue per released fix.