Feat/mempool cleanup + single EventSource #441

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

ClemensLey merged 15 commits into staging from feat/mempoolCleanup

Mar 4, 2026

package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

packages/chat/src/App.tsx

-Original file line number
+Diff line change
@@ Expand Up / @@ -22,6 +22,31 @@ export default function App() { @@
         initFlowbite()
       }, [])
+      useEffect(() => {
+        let close: (() => void) | undefined
+        const streamToComputer = async () => {
+          // Prevent double subscription in StrictMode dev
+          if (close) return
+          close = await computer.streamMempoolCleanup(async ({ revs }) => {
+            console.log('mempool cleanup txs:', revs.join(', '))
+            window.location.reload() // Reload page on mempool cleanup
+          })
+        }
+        streamToComputer()
+        return () => {
+          if (close) {
+            console.log('closing mempool cleanup SSE connection')
+            close()
+            close = undefined
+          }
+        }
+      }, [computer])
       return (
         <BrowserRouter>
           <span className="bg-gray-900/50 dark:bg-gray-900/80 z-30 inset-0 sr-only"></span>
@@ Expand Down @@

packages/chat/src/components/Chat.tsx

-Original file line number
+Diff line change
@@ Expand Up / @@ -311,6 +311,27 @@ export function Chat({ chatId }: { chatId: string }) { @@
         fetch()
       }, [computer, id, chatId, location, navigate])
+      useEffect(() => {
+        let unsubscribe: () => void
+        const subscribeToChatUpdates = async () => {
+          if (!id) return
+          unsubscribe = await computer.subscribe(
+            id, // Subscribe to this chat's ID (gets updates to its lineage)
+            // eslint-disable-next-line no-empty-pattern
+            async ({}) => {
+              // If new post, refresh the chat for now
+              await refreshChat()
+            },
+            (error) => console.error('Chat SSE error:', error),
+          )
+        }
+        subscribeToChatUpdates()
+        return () => {
+          if (unsubscribe) unsubscribe()
+        }
+      }, [id, computer])
       return (
         <>
           <div className="grid grid-cols-1 gap-4 max-w" style={{ maxHeight: 'calc(100vh - 10vh)' }}>
@@ Expand Down @@

packages/chat/src/components/Chats.tsx

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -85,26 +85,60 @@ export function Chats() {
  
      const [chatId] = useState(params.id || '')

      const [chats, setChats] = useState<ChatSc[]>([])

      const fetchChats = async () => {

        const result = await computer.getOUTXOs({ mod: VITE_CHAT_MOD_SPEC, publicKey })

        const chatsPromise: Promise<ChatSc>[] = []

        result.forEach((rev: string) => {

          chatsPromise.push(computer.sync(rev) as Promise<ChatSc>)

        })

        Promise.allSettled(chatsPromise).then((results) => {

          const successfulChats = results

            .filter((result): result is PromiseFulfilledResult<ChatSc> => result.status === 'fulfilled')

            .map((result) => result.value)

          setChats(successfulChats)

        })

      }

      useEffect(() => {

        const fetch = async () => {

          const result = await computer.getOUTXOs({ mod: VITE_CHAT_MOD_SPEC, publicKey })

          const chatsPromise: Promise<ChatSc>[] = []

          result.forEach((rev: string) => {

            chatsPromise.push(computer.sync(rev) as Promise<ChatSc>)

          })

        fetchChats()

      }, [computer, location, navigate])

          Promise.allSettled(chatsPromise).then((results) => {

            const successfulChats = results

              .filter(

                (result): result is PromiseFulfilledResult<ChatSc> => result.status === 'fulfilled',

              )

              .map((result) => result.value)

      useEffect(() => {

        let unsubscribe: () => void

        const subscribeToNewChats = async () => {

          // Stream all new chats with same mod for now

          unsubscribe = await computer.streamTXOs(

            { mod: VITE_CHAT_MOD_SPEC },

            async ({ rev }) => {

              const newChat = (await computer.sync(rev)) as ChatSc

              // Filter by ownership

              if (newChat._owners.includes(publicKey)) {

                setChats((prev) => {

                  // We receive notifications for any updates of mod

                  // We need to check if is this a creation or an update of an existing chat

                  const exists = prev.findIndex((chat) => chat._id === newChat._id)

                  if (exists !== -1) {

                    // Update existing chat with latest data

                    const updated = [...prev]

                    updated[exists] = newChat

                    return updated

                  } else {

                    // Append if truly new

                    return [...prev, newChat]

                  }

                })

              }

            },

            (error) => console.error('Chat list SSE error:', error),

          )

        }

        subscribeToNewChats()

            setChats(successfulChats)

          })

        return () => {

          if (unsubscribe) unsubscribe()

        }

        fetch()

      }, [computer, location, navigate])

      }, [computer, publicKey])

      const newChat = () => {

        Modal.showModal(newChatModal)

packages/docs/Lib/Computer/stream.md

-Original file line number
+Diff line change
@@ Expand Up / @@ -13,7 +13,7 @@ sequenceDiagram @@
         Note over C,S: Tx confirms in block
         S->>C: onMessage({rev, hex}) // Confirmation
         C->>S: unsubscribe() // Closes SSE
-    ```
+    ```
     ## Type
@@ Expand All / @@ -23,6 +23,7 @@ type StreamQuery = { @@
       asm?: string
       exp?: string
       mod?: string
+      publicKey?: string
     }
     streamTXOs(
@@ Expand All / @@ -38,20 +39,24 @@ streamTXOs( @@
     A partial `StreamQuery` object that defines the properties a transaction output must satisfy. The supported properties are:
-    | Property   | Type      | Description |
-    |------------|-----------|-------------|
-    | `satoshis` | `bigint?` | The number of satoshis on the output. |
-    | `asm`      | `string?` | The script assembly (ASM) of the output. |
-    | `exp`      | `string?` | The expression for the output's transaction. |
-    | `mod`      | `string?` | The module specifier for the output. |
+    | Property    | Type      | Description                                                 |
+    | ----------- | --------- | ----------------------------------------------------------- |
+    | `satoshis`  | `bigint?` | The number of satoshis on the output.                       |
+    | `asm`       | `string?` | The script assembly (ASM) of the output.                    |
+    | `exp`       | `string?` | The expression for the output's transaction.                |
+    | `mod`       | `string?` | The module specifier for the output.                        |
+    | `publicKey` | `string?` | Matches if the output script contains this exact publicKey. |
     Any combination of the supported fields (e.g., `mod`, `satoshis`, `asm`) can be used. All provided fields are combined with AND logic.
+    The `publicKey` option can also be used to filter for an specific script chunk. A chunk can be an opcode ("OP_1", "OP_RETURN", "OP_CHECKSIG", etc.) or a hexadecimal data push (e.g. a public key "02f7f0..." or OP_RETURN data). To match multiple chunks (e.g. both "OP_1" and a specific public key), subscribe twice with the same callback (see example below).
     Note that if a transaction has multiple outputs that encode objects you will receive updates for each of these outputs when subscribing to `exp`. This is because the expression `exp` is a property of the transaction, not of the output.
     #### `onMessage`
     The function to call when a matching transaction is detected. The callback receives an object containing:
     - `rev`: the revision of matching output, that is `<transaction id>:<output number>`
     - `hex`: the transaction hex
@@ Expand All @@
     The `streamTXOs` method enables real-time updates via Server-Sent Events (SSEs) that match a custom filter. It opens an SSE connection to the server and invokes the provided callback whenever a transaction containing an output that exactly matches all conditions in the filter is broadcast — first in the mempool and again when confirmed in a block.
+    The `publicKey` field gives powerful flexibility: you can subscribe to any individual piece of a script (opcodes or data pushes), making it easy to watch for specific patterns like OP_RETURN data, particular public keys, or script fragments.
     ## Tips
     - **Reconnection:** SSEs can drop on network hiccups—use `onError` for exponential backoff retries.
     - **Filter Precision:** Start broad (e.g., just `mod`) to test, then narrow with `satoshis` or `asm` for specificity.
     - **Performance:** Limit concurrent streams; each opens a dedicated SSE.
@@ Expand Down @@

packages/docs/Lib/Computer/streamMempoolCleanup.md

-Original file line number
+Diff line change
@@ -0,0 +1,55 @@
+    # streamMempoolCleanup
+    Subscribe to real-time mempool cleanup events. Get notified when the node removes stale unconfirmed transactions from its database, allowing apps to refresh or reload components for consistency. Powered by SSE for low-latency global notifications.
+    ```mermaid
+    sequenceDiagram
+        participant C as Client
+        participant S as Server
+        C->>S: streamMempoolCleanup(callbacks)
+        S-->>C: SSE Connected (Promise resolves)
+        Note over C,S: Mempool cleanup triggers
+        S->>C: onMessage({type: 'cleanup', staleTxIds, revs}) // Deletions
+        C->>S: unsubscribe() // Closes SSE
+    ```
+    ## Type
+    ```ts
+    streamMempoolCleanup(
+      onMessage: (event: { revs: string[] }) => void,
+      onError?: (error: Event) => void,
+    ): Promise<() => void>
+    ```
+    ### Parameters
+    #### `onMessage`
+    The function to call when a mempool cleanup event occurs. The callback receives an object containing:
+    - `revs`: Array of revisions (e.g., <txId>:<outputIndex>) affected by the cleanup.
+    #### `onError` (optional)
+    A callback invoked when an error occurs on the SSE connection, such as network interruptions or parsing failures. It receives a standard browser [`Event`](https://developer.mozilla.org/en-US/docs/Web/API/Event) object (e.g., with `type: 'error'` for connection issues).
+    For reconnection strategies, consider exponential backoff in your handler. See the [MDN SSE error handling guide](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#handling_errors) for more details.
+    ### Return Value
+    A promise that resolves to a cleanup function once the SSE connection is established. Calling the function closes the connection and stops updates.
+    ## Description
+    The `streamMempoolCleanup` method enables real-time notifications via Server-Sent Events (SSEs) for mempool cleanups. It opens a global SSE connection to the server and invokes the provided callback whenever the node performs a cleanup, removing stale unconfirmed transactions (those in the DB but not in the current mempool).
+    ## Tips
+    - **Reconnection:** SSEs can drop on network hiccups—use `onError` for exponential backoff retries.
+    - **Global Scope**: No filters; receives all cleanup events—ideal for app-wide refreshes.
+    - **Performance:** Limit concurrent streams; each opens a dedicated SSE.
+    - **Cleanup Best Practice:** Always invoke the returned function in hooks like React's `useEffect` return.

packages/docs/Node/clean-mempool.md

-Original file line number
+Diff line change
@@ -0,0 +1,69 @@
+    # clean-mempool
+    _Triggers a cleanup of stale unconfirmed mempool entries._
+    ## Description
+    Triggers a manual cleanup of the node’s internal mempool representation.
+    The Bitcoin Computer Node stores unconfirmed transactions and outputs in its database to efficiently track mempool state. Over time, some transactions may disappear from the Bitcoin node’s mempool (for example due to eviction, replacement, or restarts) without ever being confirmed.
+    This endpoint removes unconfirmed database entries that are no longer present in the Bitcoin node’s mempool, ensuring that the node’s internal state remains consistent and does not accumulate stale data.
+    It is recommended disable this endpoint for mainnet network mode, as it performs a potentially expensive database operation.
+    ## Endpoint
+    `/v1/CHAIN/NETWORK/clean-mempool`
+    ## Configuration
+    To enable this endpoint, set the following variable in your `.env` file:
+    ```bash
+    BCN_MEMPOOL_CLEANUP_ENDPOINT_ENABLED='true'
+    ```
+    If this variable is set to 'false', the endpoint will return a `403 Forbidden` response.
+    Even if disabled, the node will still perform automatic background cleanup based on the configured interval, according to the Bitcoin Node mempool `BCN_MEMPOOL_CLEANUP_INTERVAL_MS` (default is 1 hour for BTC and LTC, mainnet and testnet; and 15 minutes for DOGE and PEPE)
+    ## Example
+    ### Request
+    ```shell
+    curl -X POST http://localhost:1031/v1/LTC/regtest/clean-mempool
+    ```
+    ### Response
+    #### Success (200)
+    Returns a list of revision identifiers (`rev`) that were removed during the cleanup.
+    ```json
+    ["txid1:0", "txid2:1", "txid3:0"]
+    ```
+    Each entry represents a stale unconfirmed revision that was removed from the database.
+    #### Forbidden (403)
+    Returned when the cleanup endpoint is disabled via configuration.
+    ```
+    [
+      "Mempool cleanup is disabled"
+    ]
+    ```
+    #### Server error (500)
+    ```json
+    { "error": "Internal server error message" }
+    ```
+    ### Notes
+    - It is recommended to keep this endpoint disabled on public mainnet deployments unless explicitly required.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feat/mempool cleanup + single EventSource #441

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!