JSON RPC Endpoint: `queryGroupMessages`

[jac18281828]

Documentation for JSON RPC Endpoint: queryGroupMessages

Overview

The queryGroupMessages method is used to retrieve messages within a specified conversation. This method requires two parameters: a unique identifier for the conversation (conversationId) and the maximum number of results to return (limit).

JSON RPC Endpoint Specification

Request:

• Method: POST
• URL: /rpc/v1/queryGroupMessages
• Headers:
  • Content-Type: application/json
• Body:
  • JSON Object:
  • jsonrpc: "2.0"
  • method: "queryGroupMessages"
  • params: Array (optional parameters as required)
  • id: Request identifier (integer or string)

Method Name

queryGroupMessages

Request Parameters

1. conversationId: A unique identifier for the conversation. This is a 32-byte string, typically in hexadecimal format.
2. limit: The limit on the number of messages to return, server will enforce a maximum of 500
3. lastBlock: The last block to consider for this request, optionally empty, indicating the most recent block

Request Format

{
  "jsonrpc": "2.0",
  "method": "queryGroupMessages",
  "params": {
    "conversationId": "<conversationId>",
    "limit": "number",
    "lastBlock": "optional<number>"
  },
  "id": 1
}

• jsonrpc: Specifies the version of the JSON RPC protocol being used. Always "2.0".
• method: The name of the method being called. Here it is "rewindMessage".
• params: A structured value holding the parameters necessary for the method. It contains:
  • conversationId: The unique identifier for the conversation.
  • limit: The maximum number of messages to return, server enforced limit is 500
• id: A unique identifier established by the client that must be number or string. Used for correlating the response with the request.

Response Format

The response will include the result of the operation or an error if the operation was unsuccessful. The included array (message) will contain the messages requested. The block range covered by this operation, (startBlock –endBlock) will be included in the result. The block range may be using for paging over past results.

Success Response

{
  "jsonrpc": "2.0",
  "result": "status",
  "message": [
    "<msg 1 ... >",
    ...
  ],
  "startBlock": "block number",
  "endBlock": "block number",
  "id": 1
}

• result: Specifies success of the operation. The nature of this specification can vary based on the implementation.

Error Response

{
  "jsonrpc": "2.0",
  "error": {
    "code": <error_code>,
    "message": "<error_message>"
  },
  "id": 1
}

• error: An object containing details about the error.
  • code: A numeric error code.
  • message: A human-readable string describing the error.

Example Usage

Request

{
  "jsonrpc": "2.0",
  "method": "queryGroupMessages",
  "params": {
    "conversationId": "0x1234abcd...",
    "limit": "20"
  },
  "id": 42
}

Response

{
  "jsonrpc": "2.0",
  "result": "5 message retrieved",
  "message": [
     "abc...",
     ...
  ],
  "startBlock": 11808,
   "endBlock": 11810,
  "id": 42
}

Possible Implementation

    let mut n = limit;
    let mut rewind = MessageRewind {
        message: Vec::new(),
        last_change: U256::zero(),
    };
    let mut last_change = last_change_result.unwrap();
    rewind.last_change = last_change;
    while last_change != U256::zero() {
        tracing::debug!("prev_change: {}", last_change);
        let conversation_topic = [H256::from(conversation_id)];
        let contract_addr = SENDER_CONTRACT.parse::<Address>().unwrap();
        let filter = Filter::new()
            .from_block(U64::from(last_change.as_u64()))
            .to_block(U64::from(last_change.as_u64()))
            .event("PayloadSent(bytes32,bytes,uint256)")
            .address(vec![contract_addr])
            .topic1(conversation_topic.to_vec());
        let logs = self.client.get_logs(&filter).await;
        if let Ok(logs) = logs {
            for log in logs.iter() {
                if tracing::level_enabled!(tracing::Level::TRACE) {
                    tracing::trace!("log: {:?}", log);
                }
                let param_result = abi_decode_payload_sent(log.data.to_vec());
                if let Ok(param) = param_result {
                    tracing::debug!("param: {:?}", param);
                    let message = param[0].clone().into_string().unwrap();
                    if tracing::level_enabled!(tracing::Level::TRACE) {
                        tracing::trace!("message: {message}");
                    }
                    rewind.message.push(message);
                    last_change = param[1].clone().into_uint().unwrap();
                } else {
                    let err = param_result.unwrap_err();
                    tracing::error!("param error: {:?}", err);
                    return Err(err);
                }

                n -= 1;
                if n == 0 {
                    last_change = U256::zero();
                    break;
                }
            }
        }
    }

    rewind.message.reverse();
    tracing::info!("{} messages found", rewind.message.len());
    Ok(rewind)

Notes

• Ensure that the conversationId is valid and exists
• method is public, does not require authentication
• The system should have proper error handling to deal with invalid requests, unauthorized access, and other potential issues.

[jac18281828]

Reference implementation: https://github.com/xmtp/xps-conversation-producer