DataStore subscription does not work for only a few models.

Description

Hi everyone, I'm using DataStore subscriptions in my project to enable real-time chat for users. Initially, messages were not arriving in real-time, but after deleting and re-adding the conversation and message models from the database, neither conversations nor messages are being delivered in real-time anymore.(While it was not working just for the message before, deleting the entire database and adding it again solved my problem.) During this time, I can see that messages and conversations are being created in the database in real-time through Amplify Studio. I am not getting any errors within the application. But if i use GraphQL subscription to do that, it works correctly. What could be the reason for this? I am adding video recording and screenshot for understanding easily.

https://github.com/user-attachments/assets/e531949a-84ed-4e21-ad66-ccf4f60508cd

Database_screen

Steps to Reproduce

I am adding the code I wrote for the conversation subscription, and I am using a similar code for the message (I am using riverpod to do that. i also tried to use it like in Amplify Documentation but result doesn't change.). conversation_in_api conversation_in_controller conversation_stream_code_in_build

Screenshots

https://github.com/user-attachments/assets/c7a41e76-959f-45a4-97c2-4b4291d8fd93

conversation_in_api conversation_in_controller conversation_stream_code_in_build Database_screen

Platforms

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

Flutter Version

3.22.3

Amplify Flutter Version

2.3.0

Deployment Method

Amplify CLI

Schema

type Message @model @auth(rules: [{allow: private, operations: [create, read]}, {allow: owner, operations: [create, read, update]}]) {
  id: ID!
  text: String!
  visualDatas: [String]!
  timestamp: AWSDateTime!
  conversationID: ID! @index(name: "byConversation")
  userID: ID! @index(name: "byUser")
}

type Conversation @model @auth(rules: [{allow: private, operations: [create, read]}]) {
  id: ID!
  members: [String]!
  Messages: [Message] @hasMany(indexName: "byConversation", fields: ["id"])
}

type HuCodeNotification @model @auth(rules: [{allow: private, operations: [create, read]}]) {
  id: ID!
  content: String!
  isRead: Boolean!
  notificationType: String!
  receiverUserID: ID! @index(name: "byUser")
}

enum FollowRequestStatus {
  PENDING
  ACCEPTED
  REJECTED
}

type FollowRequest @model @auth(rules: [{allow: private}]) {
  id: ID!
  senderUserID: String!
  receiverUserID: String!
  status: FollowRequestStatus
}

type Report @model @auth(rules: [{allow: private, operations: [create, read]}, {allow: owner, operations: [create, read, update]}]) {
  id: ID!
  reasonExplanation: String!
  reason: [String]!
  reportedContentID: String!
  reportType: String!
  userID: ID! @index(name: "byUser")
}

type Reason @model @auth(rules: [{allow: private, operations: [create, read]}, {allow: owner}]) {
  id: ID!
  choice: String!
  reasonText: String!
  likes: [String]!
  dislikes: [String]!
  surveyID: ID! @index(name: "bySurvey")
  userID: ID! @index(name: "byUser")
}

type Survey @model @auth(rules: [{allow: private, operations: [create, read]}]) {
  id: ID!
  question: String!
  options: [String]!
  endDate: AWSDateTime!
  Reasons: [Reason] @hasMany(indexName: "bySurvey", fields: ["id"])
}

type Post @model @auth(rules: [{allow: private, operations: [create, read]}, {allow: owner}]) {
  id: ID!
  text: String!
  views: Int!
  dislikes: [String]!
  likes: [String]!
  comments: [String]!
  commentedPost: String
  legislationTags: [String]!
  quotedPost: String
  hashtags: [String]!
  whoCanSee: String!
  lawTags: [String]!
  visualDatas: [String]!
  userID: ID! @index(name: "byUser")
}

type University @model @auth(rules: [{allow: private, operations: [create, read]}, {allow: owner}]) {
  id: ID!
  name: String!
  degree: String!
  gpa: Float!
  userID: ID! @index(name: "byUser")
}

type User @model @auth(rules: [{allow: private, operations: [create, read]}, {allow: owner}]) {
  id: ID!
  userType: String!
  name: String!
  surname: String!
  phoneNumber: String!
  email: String!
  birthdate: AWSDate!
  gender: String
  barAssociation: String
  barAssociationNumber: Int
  tbbRegistrationNumber: Int
  interests: [String]!
  followers: [String]!
  followed: [String]!
  aboutMe: String
  Address: String
  visiblePhoneNumbers: [String]
  visibleEmail: String
  bannerPicture: String
  profilePicture: String
  identityFile: String
  barAssociationfile: String
  studentIdentityFile: String
  confirmedUser: Boolean!
  bannedPostIDs: [String]!
  Universities: [Message] @hasMany(indexName: "byUser", fields: ["id"])
  Posts: [Message] @hasMany(indexName: "byUser", fields: ["id"])
  bannedUserIDs: [String]!
  Reasons: [Message] @hasMany(indexName: "byUser", fields: ["id"])
  Reports: [Message] @hasMany(indexName: "byUser", fields: ["id"])
  Notifications: [Message] @hasMany(indexName: "byUser", fields: ["id"])
  Messages: [Message] @hasMany(indexName: "byUser", fields: ["id"])
}

Hi @BatuhancanG, thank you for submitting this issue, we will take a look at this issue and get back to you with any updates or questions.

@BatuhancanG I'm investigating this issue and while working to build the data backend with CLI Gen 2 I faced some issue with the data model. to be able to deploy the data backend successfully I needed to make the following updates:

update the HuCodeNotification model to Notification
update the User model field's type for these fields: Universities, Posts, Reasons, Reports, Notifications, Messages.

here the Gen 2 data model that I have and was able to deploy it successfully.

const schema = a.schema({
  Message: a
    .model({
      text: a.string().required(),
      visualDatas : a.string().array().required(),
      timestamp: a.datetime().required(),
      conversationID: a.id().required(),
      conversation: a.belongsTo('Conversation', 'conversationID'),
      userID: a.id().required(),
      user: a.belongsTo('User', 'userID'),
    })
    .authorization((allow) => [allow.authenticated().to(['create', 'read']),allow.owner().to(['create', 'read', 'update'])]),

  Conversation: a
    .model({
      members: a.string().array().required(),
      Messages: a.hasMany('Message', 'conversationID'),
    })
    .authorization((allow)=> [allow.authenticated().to(['create','read'])]),

  Notification:a
    .model({
      content: a.string().required(),
      isRead: a.boolean().required(),
      notificationType: a.string().required(),
      receiverUserID: a.id().required(),
      receiverUser: a.belongsTo('User', 'receiverUserID'),
    })
    .authorization((allow)=> [allow.authenticated().to(['create','read'])]),

  Report: a
    .model({
      reasonExplanation: a.string().required(),
      reason: a.string().array().required(),
      reportedContentID: a.string().required(),
      reportType: a.string().required(),
      userID: a.id().required(),
      user: a.belongsTo('User', 'userID'),
    })
    .authorization((allow) => [allow.authenticated().to(['create', 'read']),allow.owner().to(['create', 'read', 'update'])]),

  Reason:a
    .model({
      choice: a.string().required(),
      reasonText: a.string().required(),
      likes: a.string().array().required(),
      dislikes: a.string().array().required(),
      surveyID: a.id().required(),
      survey: a.belongsTo('Survey', 'surveyID'),
      userID: a.id().required(),
      user: a.belongsTo('User', 'userID'),
    })
    .authorization((allow) => [allow.authenticated().to(['create', 'read']),allow.owner()]),

  Survey:a
    .model({
      question: a.string().required(),
      options: a.string().array().required(),
      endDate: a.datetime().required(),
      Reasons: a.hasMany('Reason','surveyID'),
    })      
    .authorization((allow) => [allow.authenticated().to(['create', 'read'])]),

  Post:a
    .model({
      text: a.string().required(),
    views: a.integer().required(),
    dislikes: a.string().array().required(),
    likes: a.string().array().required(),
    comments: a.string().array().required(),
    commentedPost: a.string(),
    legislationTags: a.string().array().required(),
    quotedPost: a.string(),
    hashtags: a.string().array().required(),
    whoCanSee: a.string().required(),
    lawTags: a.string().array().required(),
    visualDatas: a.string().array().required(),
    userID: a.id().required(),
    user: a.belongsTo('User', 'userID'),
    })      
    .authorization((allow) => [allow.authenticated().to(['create', 'read']),allow.owner()]),

  University:a
    .model({
      name: a.string().required(),
      degree: a.string().required(),
      gpa: a.float().required(),
      userID: a.id().required(),
      user: a.belongsTo('User', 'userID'),
    })
    .authorization((allow) => [allow.authenticated().to(['create', 'read']),allow.owner()]),

  User:a
    .model({
      userType: a.string().required(),
      name: a.string().required(),
      surname: a.string().required(),
      phoneNumber: a.string().required(),
      email: a.string().required(),
      birthdate: a.date().required(),
      gender: a.string(),
      barAssociation: a.string(),
      barAssociationNumber: a.integer(),
      tbbRegistrationNumber: a.integer(),
      interests: a.string().array().required(),
      followers: a.string().array().required(),
      followed: a.string().array().required(),
      aboutMe: a.string(),
      Address: a.string(),
      visiblePhoneNumbers: a.string().array(),
      visibleEmail: a.string(),
      bannerPicture: a.string(),
      profilePicture: a.string(),
      identityFile: a.string(),
      barAssociationfile: a.string(),
      studentIdentityFile: a.string(),
      confirmedUser: a.boolean().required(),
      bannedPostIDs: a.string().array().required(),
      Universities: a.hasMany('University', 'userID'),
      Posts: a.hasMany('Post','userID'),
      bannedUserIDs: a.string().array().required(),
      Reasons: a.hasMany('Reason', 'userID'),
      Reports: a.hasMany('Report','userID'),
      Notifications: a.hasMany('Notification','receiverUserID'),
      Messages: a.hasMany('Message', 'userID'),
    })
    .authorization((allow) => [allow.authenticated().to(['create', 'read']),allow.owner()]),      
});

While we are investigating this issue would you please confirm that the data model you provided is up to date and you were able to deploy your data backend successfully using the provided data model?

Hi, thank you for your response. I’m not using Gen2. I’m developing my project with Gen1. I don’t know if this makes a difference, but I wanted to mention it. Also, the relations in my User model were correct initially, but whenever I make changes to the database, the relations break. It’s only the relations with the User model that get broken. I think this is a other issue. I recorded a video to help you understand the problem better, and I’m attaching it as well. As you suggested, I changed HucodeNotification to Notification and fixed the relations in the user model, but the problem still persists. This issue only occurs with the Message part. It doesn’t happen with Conversation or with subscriptions for other models. I’ve recorded a video to show the situation as well, where you can see that all the messages are saved to the database, but the other party doesn’t receive them in real-time. Additionally, after changing the name of Notification, I started getting an error in Amplify Studio. I'm attaching a screenshot of the error as well.

https://github.com/user-attachments/assets/96079c24-25c1-4dbf-a2c4-3fddf6f31deb

https://github.com/user-attachments/assets/541b77bf-9efa-41a0-92c5-8d5cd91bae7f

Ekran görüntüsü 2024-08-24 152147

"When I check the overall application right now, the subscription doesn't work even when the user's data is updated in real-time. The real-time updates to user information don't work when using Amplify.DataStore.observeQuery, but when I use Amplify.API.subscribe, I can receive all real-time updates to user information. I'm also attaching screenshots of the code I wrote for your understanding." userAPI_observeQuery controllerObserveQuery api_userAPI apicontroller api_screen

We don't want to use the API.subscribe method because, in the social media application we are developing, we don't want to create new requests and pull the latest data from the database every time there's a page change. We believe this would be very costly for us. Therefore, it's important for us to build our application using DataStore. We would appreciate it if you could assist us as soon as possible.

Hello @BatuhancanG. You would not necessarily need to make a new request on each change. You could do the following instead:

Make an initial query to obtain initial data. Use this data as needed and store it in-memory.
Subscribe to updates on the model and mutate the in-memory list from step 1 as appropriate
- For onCreate events, add the item to the list of models
- For onDelete events, remove the item from the list of models
- For onUpdate events, update the item in the list (or remove it if it no longer meets the criteria of your query)

This is how observeQuery works under the hood.

If there were an observeQuery like API available in the API category, would the API category meet your needs?

We do have an open feature request for an observeQuery like API: https://github.com/aws-amplify/amplify-flutter/issues/2414

Feel free to give the issue a 👍 and leave a comment with your use case. This helps us prioritize issues and requests.

Hello @BatuhancanG. You would not necessarily need to make a new request on each change. You could do the following instead:

Make an initial query to obtain initial data. Use this data as needed and store it in-memory.

Subscribe to updates on the model and mutate the in-memory list from step 1 as appropriate

For onCreate events, add the item to the list of models

For onDelete events, remove the item from the list of models

For onUpdate events, update the item in the list (or remove it if it no longer meets the criteria of your query)

This is how observeQuery works under the hood.

If there were an observeQuery like API available in the API category, would the API category meet your needs?

Hi, thank you for your answer. In the method you suggested, I would need to create a database on the phone's memory after pulling the data from my database, for example, by using a library like sqflite or Hive. This would create an additional workload for me. DataStore, on the other hand, was saving me from this issue. After all, even if I wrote a stateNotifier with Riverpod just to store the data in phone memory, I would have to make another query every time the app is closed and reopened. Making a query every time the app is closed and reopened, even for unchanged data, doesn't make much sense to me. In the other method, I would necessarily need a database on the phone's memory, like SQLite or Hive. This would be an extra workload for me, which is why I'm not keen on using those options. If I can get real-time updates with DataStore, my current problems would be solved. Additionally, it doesn't make sense to subscribe separately to onCreate, onDelete, and onUpdate in the API. I also think it's more logical to have a single subscription like observeQuery that covers all of them.

I would have to make another query every time the app is closed and reopened DataStore will do a sync each time the app is closed an opened to receive events that were missed while the app was closed so these queries are happening regardless.

I understand that it is extra workload. If an API existed within the API category (similar to observeQuery) that removed the need to maintain 3 separate subscriptions and make the initial query each time the app was opened, would that remove your need for DataStore?

DataStore is an offline first solution. A typical use case for offline first would be an app where users of the app are commonly offline for multiple days at a time and need to maintain the ability to read and write data. If you do not have a use case that requires offline first, we typically recommend using GraphQL API directly as it is more flexible.

I think I won't be using DataStore because the subscribe feature is not working correctly, and I assume you don't have a solution for this problem either. I will replace all the DataStore sections in my application with API. Yes, using a single method like observeQuery in DataStore to utilize onDelete, onUpdate, and onCreate features is a great advantage. However, when I start using API, there's another feature that DataStore has, but API doesn't: the sort feature. In my application, I use code like this. I sort sent messages by the send time in descending order using the sortBy parameter and fetch them 10 at a time. This way, if the user wants to see older messages, they only appear then. But right now, I can't do this. There is no such sorting feature in “ModelQueries.list.” Even if I set the limit to 10, it returns 10 messages in a mixed order. So, how can I overcome this issue? Also, pagination is quite complicated compared to DataStore. Can you provide a solution with a sample code for my problem? apisonmesaj datastoresonmesaj

Hello, can someone help me? Nobody has gotten back to me yet.

@BatuhancanG we have an open feature request for GraphQL API support sorting by secondary index, #4942. in your case can you use a query predicate on the Message timestamp as an alternative solution? (I understand it does not satisfay the limit=10 that you can do with the DataStore)

aws-amplify / amplify-flutter