Adding psf-memo-indexer

This commit is contained in:
Chris Troutner
2026-08-26 09:30:44 -07:00
parent 28bde991b5
commit 5a68a42f77
67 changed files with 13681 additions and 1 deletions
+1 -1
View File
@@ -7,6 +7,6 @@ leveldb/
logs/
.env
coverage/
data/
.gitsigners
+13
View File
@@ -0,0 +1,13 @@
PSF_MEMO_DB_URL=http://localhost:5021
RPC_IP=172.17.0.1
RPC_PORT=8332
RPC_USER=bitcoin
RPC_PASS=password
ZMQ_PORT=28332
TX_REST_API_PORT=5455
TX_REST_API_IP=localhost
START_BLOCK_HEIGHT=525000
SEEN_TX_MAX=100000
FILTER_CONCURRENCY=20
MEMO_TX_CONCURRENCY=20
DEBUG_LEVEL=0
+5
View File
@@ -0,0 +1,5 @@
node_modules/
.env
coverage/
logs/
production/data/
+17
View File
@@ -0,0 +1,17 @@
GNU GENERAL PUBLIC LICENSE
Version 3, 29 June 2007
Copyright (C) 2026 Permissionless Software Foundation
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
+240
View File
@@ -0,0 +1,240 @@
# psf-memo-indexer
Indexes [Memo protocol](https://memo.cash) transactions on Bitcoin Cash. Architecture mirrors [psf-slp-indexer-g2](https://github.com/Permissionless-Software-Foundation/psf-slp-indexer-g2).
## Overview
Two processes:
- **Block indexer** — IBD from block 525000, then ZMQ new-block processing
- **TX indexer** — mempool transactions via ZMQ after IBD signals `/tx-start`
Data is stored in [psf-memo-db](../psf-memo-db) via REST.
## Developer documentation
Architecture, theory of operation, and design tradeoffs: [dev-docs/](./dev-docs/README.md).
## Requirements
- node ^20
- npm ^10
- BCH full node (RPC + ZMQ)
- Running psf-memo-db
## Installation
```bash
cd psf-memo-indexer
npm install
cp .env-example .env
```
## Usage
Start the database:
```bash
cd ../psf-memo-db && npm start
```
Block indexer:
```bash
npm run block-indexer
```
TX indexer (separate terminal):
```bash
npm run tx-indexer
```
## Configuration
See `.env-example`. Key variables:
| Variable | Default | Description |
|----------|---------|-------------|
| `PSF_MEMO_DB_URL` | `http://localhost:5021` | psf-memo-db URL |
| `START_BLOCK_HEIGHT` | `525000` | First block to index |
| `RPC_IP` / `RPC_PORT` | `172.17.0.1` / `8332` | Full node RPC |
| `ZMQ_PORT` | `28332` | Full node ZMQ |
| `TX_REST_API_PORT` | `5455` | TX indexer control API |
| `FILTER_CONCURRENCY` | `20` | Parallel Memo tx detection per block |
| `MEMO_TX_CONCURRENCY` | `20` | Parallel Memo tx processing per block |
| `DEBUG_LEVEL` | `0` | `0` = block summary only; `1` = log each Memo tx action type and success/failure |
## Tests
```bash
npm test
```
## Production (Docker)
Docker Compose under [production/docker](./production/docker) runs the full stack. Images clone their source from GitHub at build time (same pattern as [psf-slp-indexer-g2](https://github.com/Permissionless-Software-Foundation/psf-slp-indexer-g2)).
### Services
| Service | Container | Host port | Purpose |
|---------|-----------|-----------|---------|
| `memo-db` | `memo-db` | `5021` | LevelDB REST API ([psf-memo-db](https://github.com/Permissionless-Software-Foundation/psf-memo-db)) |
| `block-indexer` | `memo-block-indexer` | — | IBD + ZMQ block indexing |
| `tx-indexer` | `memo-tx-indexer` | `5455` | Mempool TX indexing (`/tx-start` control API) |
| `memo-client` | `memo-client` | `3000` | React SPA ([psf-memo-client](https://github.com/Permissionless-Software-Foundation/psf-memo-client)), nginx |
LevelDB data persists on the host at `production/data/leveldb`.
### Prerequisites
- Docker Engine and Docker Compose v2 (`docker compose`)
- A Bitcoin Cash full node with RPC and ZMQ reachable from the containers
- On a typical Linux Docker host, the bridge gateway `172.17.0.1` reaches services on the host (RPC, ZMQ, and sibling containers published on host ports)
### 1. Configure environment files
```bash
cd production/docker
cp memo-db/.env-example memo-db/.env
cp block-indexer/.env-example block-indexer/.env
cp tx-indexer/.env-example tx-indexer/.env
cp memo-client/.env-example memo-client/.env
```
Edit each `.env` before building or starting.
#### `memo-db/.env`
| Variable | Typical value | Description |
|----------|---------------|-------------|
| `PORT` | `5021` | REST API listen port |
| `SVC_ENV` | `prod` | Runtime environment |
| `BACKUP_QTY` | `3` | How many epoch zip backups to keep |
| `EXIT_ON_MISSING_BACKUP` | `false` | Exit if expected backup is missing |
#### `block-indexer/.env` and `tx-indexer/.env`
| Variable | Typical Docker value | Description |
|----------|----------------------|-------------|
| `PSF_MEMO_DB_URL` | `http://172.17.0.1:5021` | URL of `memo-db` from inside the container |
| `RPC_IP` / `RPC_PORT` | `172.17.0.1` / `8332` | BCH full node RPC |
| `ZMQ_PORT` | `28332` | BCH full node ZMQ |
| `RPC_USER` / `RPC_PASS` | *(your node auth)* | RPC credentials |
| `TX_REST_API_PORT` | `5455` | TX indexer HTTP port |
| `TX_REST_API_IP` | `172.17.0.1` | Where the block indexer reaches the TX indexer |
| `START_BLOCK_HEIGHT` | `525000` | First block (block indexer only) |
| `FILTER_CONCURRENCY` / `MEMO_TX_CONCURRENCY` | `20` | Parallelism (block indexer) |
| `DEBUG_LEVEL` | `0` | Block-indexer log verbosity |
| `SEEN_TX_MAX` | `100000` | TX indexer seen-tx cache size |
Use a hostname or IP your containers can actually reach for RPC, ZMQ, and `memo-db`. `172.17.0.1` is the usual Docker bridge address when those services are published on the host.
#### `memo-client/.env`
| Variable | Example | Description |
|----------|---------|-------------|
| `REACT_APP_MEMO_DB_URL` | `http://localhost:5021` | Browser-facing base URL of `memo-db` (no trailing slash) |
This value is **baked into the SPA at image build time**. Create React App reads it from `memo-client/.env` during `npm run build`.
- Local / same-machine browser: `http://localhost:5021` (or `http://<host-ip>:5021`)
- Separate domains: `https://api.mydomain.com` when the client is at `https://client.mydomain.com`
Changing `REACT_APP_MEMO_DB_URL` requires rebuilding the `memo-client` image (see below).
### 2. Build images
```bash
cd production/docker
docker compose build
```
Rebuild a single service after changing its Dockerfile or (for the client) `.env`:
```bash
docker compose build --no-cache memo-client
docker compose build block-indexer
docker compose build tx-indexer
docker compose build memo-db
```
`block-indexer` and `tx-indexer` use explicit image names (`memo-block-indexer`, `memo-tx-indexer`) so they do not collide with similarly named images from other projects (for example `psf-slp-indexer-g2`).
### 3. Start the stack
Preferred order: database first, then indexers, then the client.
```bash
cd production/docker
docker compose up -d memo-db
docker compose up -d block-indexer tx-indexer
docker compose up -d memo-client
```
Or start everything at once:
```bash
docker compose up -d
```
### 4. Verify
| Check | URL / command |
|-------|----------------|
| Database API / docs | http://localhost:5021/ |
| Database health | http://localhost:5021/health |
| TX indexer | http://localhost:5455/ (control API; `/tx-start` after IBD) |
| Front-end client | http://localhost:3000/ |
```bash
docker compose ps
docker compose logs -f memo-db
docker compose logs -f block-indexer
docker compose logs -f tx-indexer
docker compose logs -f memo-client
```
### 5. Day-to-day operations
```bash
# Stop all services
docker compose down
# Restart one service
docker compose restart block-indexer
# Rebuild and recreate after config or image changes
docker compose up -d --build memo-client
```
Start scripts under each service directory (`start-*.sh`) and `.env` files are bind-mounted into the containers. Edit them on the host and restart the service (no rebuild) unless you changed something that only applies at image build time (notably `memo-client/.env`).
### Production domains example
When the client and API are on different hostnames:
1. Set `memo-client/.env`:
```bash
REACT_APP_MEMO_DB_URL=https://api.mydomain.com
```
2. Rebuild and redeploy the client:
```bash
docker compose build --no-cache memo-client
docker compose up -d memo-client
```
3. Reverse-proxy:
- `client.mydomain.com` → host port `3000` (`memo-client`)
- `api.mydomain.com` → host port `5021` (`memo-db`)
`memo-db` enables CORS with `origin: '*'`, so the browser may call the API from another subdomain once HTTPS and DNS are in place.
## License
GPL v3
+35
View File
@@ -0,0 +1,35 @@
import 'dotenv/config'
export default {
psfMemoDbUrl: process.env.PSF_MEMO_DB_URL || 'http://localhost:5021',
rpcIp: process.env.RPC_IP || '172.17.0.1',
rpcPort: process.env.RPC_PORT || '8332',
zmqPort: process.env.ZMQ_PORT || '28332',
rpcUser: process.env.RPC_USER || 'bitcoin',
rpcPass: process.env.RPC_PASS || 'password',
txRestApiPort: process.env.TX_REST_API_PORT ? parseInt(process.env.TX_REST_API_PORT) : 5455,
txRestApiIp: process.env.TX_REST_API_IP || 'localhost',
seenTxMax: process.env.SEEN_TX_MAX ? parseInt(process.env.SEEN_TX_MAX) : 100000,
zmqTxQueueMax: process.env.ZMQ_TX_QUEUE_MAX ? parseInt(process.env.ZMQ_TX_QUEUE_MAX) : 50000,
zmqBlockQueueMax: process.env.ZMQ_BLOCK_QUEUE_MAX ? parseInt(process.env.ZMQ_BLOCK_QUEUE_MAX) : 1000,
txCacheMax: process.env.TX_CACHE_MAX ? parseInt(process.env.TX_CACHE_MAX) : 100000,
filterConcurrency: process.env.FILTER_CONCURRENCY
? parseInt(process.env.FILTER_CONCURRENCY)
: 20,
memoTxConcurrency: process.env.MEMO_TX_CONCURRENCY
? parseInt(process.env.MEMO_TX_CONCURRENCY)
: 20,
startBlockHeight: process.env.START_BLOCK_HEIGHT
? parseInt(process.env.START_BLOCK_HEIGHT)
: 525000,
exitOnMissingBackup: process.env.EXIT_ON_MISSING_BACKUP === 'true',
debugLevel: process.env.DEBUG_LEVEL !== undefined
? parseInt(process.env.DEBUG_LEVEL, 10)
: 0
}
+24
View File
@@ -0,0 +1,24 @@
# Memo Indexer Developer Documentation
This directory documents the architecture, operation, and design rationale for the Permissionless Software Foundation Memo indexing stack:
| Repository | Role |
|------------|------|
| [psf-memo-db](../../psf-memo-db) | LevelDB persistence exposed as a REST API |
| [psf-memo-indexer](../) | Block and mempool indexer (this repo) |
Protocol reference: [memo-protocol.md](../../memo-protocol.md). Reference implementation: [memo/index](../../index) (Go).
## Reading order
1. **[Overview](./overview.md)** — Why these services exist and what they produce
2. **[Architecture](./architecture.md)** — Components, processes, Clean Architecture layout, data flow
3. **[Theory of operation](./theory-of-operation.md)** — IBD, ZMQ, parsing, indexing pipeline step-by-step
4. **[psf-memo-db](./psf-memo-db.md)** — Database service design and REST contract
5. **[Design decisions and tradeoffs](./design-decisions-and-tradeoffs.md)** — Comparisons to SLP and the Go indexer, scope limits, future work
## Related material
- Production deployment: [production/docker](../production/docker/)
- SLP analogue: [psf-slp-indexer-g2](https://github.com/Permissionless-Software-Foundation/psf-slp-indexer-g2) and [psf-slp-db](https://github.com/Permissionless-Software-Foundation/psf-slp-db)
- Clean Architecture primer: [Chris Troutner — Clean Architecture](https://christroutner.github.io/trouts-blog/blog/clean-architecture)
+177
View File
@@ -0,0 +1,177 @@
# Architecture
## System context
```mermaid
flowchart TB
subgraph node [BCH_full_node]
RPC[JSON-RPC]
ZMQ[ZMQ_rawtx_rawblock]
end
subgraph indexer [psf_memo_indexer]
BI[Block_indexer_process]
TI[TX_indexer_process]
BI -->|GET_tx-start| TI
end
subgraph db [psf_memo_db]
Koa[Koa_REST]
LDB[(LevelDB_instances)]
Koa --> LDB
end
RPC --> BI
RPC --> TI
ZMQ --> BI
ZMQ --> TI
BI -->|axios_level_CRUD| Koa
TI -->|axios_level_CRUD| Koa
```
## Two-process indexer design
The indexer splits work the same way as [psf-slp-indexer-g2](https://github.com/Permissionless-Software-Foundation/psf-slp-indexer-g2):
| Process | Responsibility | Why separate |
|---------|----------------|--------------|
| **Block indexer** | Historical catch-up (IBD) and confirmed blocks via ZMQ | Heavy, sequential block work; must finish before mempool indexing is trustworthy for “at tip” |
| **TX indexer** | Unconfirmed transactions via ZMQ | High volume, duplicate ZMQ events; isolated so IBD memory/CPU does not contend with mempool flood |
Coordination is deliberately minimal: after IBD, the block indexer issues `GET http://{TX_REST_API_IP}:{TX_REST_API_PORT}/tx-start`. The TX indexer polls that flag every two seconds until set, then subscribes to ZMQ.
**Tradeoff:** No shared memory or message queue between processes—only HTTP and the database. Simplicity and operational parity with SLP outweigh lower latency startup.
## Clean Architecture (indexer)
Both repos follow [Clean Architecture](https://christroutner.github.io/trouts-blog/blog/clean-architecture): dependencies point inward; framework and I/O live at the edges.
### psf-memo-indexer layers
```text
Entry points (framework)
psf-memo-block-indexer.js
psf-memo-tx-indexer.js
Controllers
keyboard.js — graceful stop (q key)
tx-rest-api.js — Express /tx-start
Use cases
index-blocks.js — processBlock, processMemoTx (parallel per block)
filter-block.js — filterMemoTxs (parallel pre-check, block order preserved)
state.js — synced block height
action-types/*.js — per Memo action handlers
utils.js
Adapters
rpc.js — full node JSON-RPC
zmq.js — @psf/bitcoincash-zmq-decoder
transaction.js — fetch tx + memo detection cache
status-db.js, *-db — axios → psf-memo-db
tx-indexer.js — start signal to TX process
backup-db.js — POST /level/backup
Libraries
memo-parser.js — OP_RETURN pushdata, signer address
memo-codes.js — action prefixes and limits
```
**Rule:** Use cases contain business rules (valid pushdata counts, reply parent links, like tips). Adapters only move bytes on the network or disk API.
### psf-memo-db layers
```text
index.js → bin/server.js (Koa)
Controllers
rest-api/index.js
rest-api/level/* — CRUD handlers
rest-api/health/
Use cases
index.js — minimal stub (parity with psf-slp-db)
Adapters
level-db.js — open/close LevelDB instances
db-backup.js — zip / restore
```
Level CRUD bypasses use cases intentionally—same as `psf-slp-db`s `/level` controller calling `adapters.level.*Db.put` directly. The DB service is a thin persistence plane, not a domain model server.
## Repository layout (indexer)
| Path | Purpose |
|------|---------|
| `config/index.js` | Environment-driven settings |
| `src/lib/` | Pure protocol parsing (no I/O) |
| `src/adapters/` | External systems |
| `src/use-cases/` | Indexing orchestration and handlers |
| `src/controllers/` | HTTP and keyboard |
| `production/docker/` | Docker Compose, per-service Dockerfile and `.env` |
| `test/unit/` | Mocha + c8 |
## Repository layout (database)
| Path | Purpose |
|------|---------|
| `bin/server.js` | Koa bootstrap (mirrors psf-slp-db) |
| `config/env/` | `SVC_ENV` profiles |
| `src/adapters/level-db.js` | Twelve LevelDB stores under `leveldb/current/` |
| `src/controllers/rest-api/level/` | Generic CRUD + status + backup routes |
| `leveldb/current/{name}/` | Runtime data (gitignored) |
| `leveldb/zips/` | Epoch backups |
## LevelDB schema (summary)
Each store is a separate LevelDB with JSON values. Keys are txids, addresses, or composite strings depending on entity. Full detail: [psf-memo-db.md](./psf-memo-db.md).
| Store | Typical key | Value role |
|-------|-------------|------------|
| `status` | `status` | `syncedBlockHeight`, `chainBlockHeight`, `startBlockHeight` |
| `posts` | txid | Author address, text, timestamp |
| `postParents` | child txid | Parent txid (reply) |
| `postChildren` | `parentTxid:childTxid` | One reply link per parentchild pair |
| `likes` | like txid | Post txid, liker, optional tip |
| `names` | address | Display name + provenance txid |
| `profiles` | address | Profile text |
| `profilePics` | address | Avatar URL |
| `follows` | `follower:followeePkHash` | Follow/unfollow event |
| `rooms` | composite | Topic posts and topic follows |
| `processErrors` | txid | Validation / parse failures |
| `ptxs` | txid | Idempotency marker (already processed) |
## External dependencies
| Package | Used by | Role |
|---------|---------|------|
| `@psf/bch-js` | Indexer | Script decompile for signer address; optional REST |
| `@psf/bitcoincash-zmq-decoder` | Indexer | Decode `rawtx` / `rawblock` ZMQ messages |
| `zeromq` | Indexer | Subscribe to full node |
| `@chris.troutner/retry-queue` | Indexer | Retry RPC on transient failure |
| `axios` | Indexer | psf-memo-db REST client |
| `p-queue` / `p-retry` | Indexer | Parallel block filter and parallel per-tx processing within a block |
| `express` | Indexer | TX indexer control API |
| `level` | psf-memo-db | Embedded JSON LevelDB |
| `koa` + `koa-router` | psf-memo-db | REST server |
**Not used:** `slp-parser`, Lokad ID checks, DAG sorting, token UTXO graphs.
## Deployment topology
```text
production/docker/
├── docker-compose.yml
├── memo-db/ → build context: ../../psf-memo-db
├── block-indexer/ → build context: indexer repo root
└── tx-indexer/
```
Volumes mount `.env` and start scripts per service, matching the SLP production pattern. LevelDB data can persist under `production/data/leveldb`.
@@ -0,0 +1,191 @@
# Design Decisions and Tradeoffs
This document records **why** the Memo indexer stack looks the way it does, including deliberate limitations and comparisons to adjacent systems.
## 1. Mirror the SLP indexer architecture
**Decision:** Two processes (block + TX), separate `psf-memo-db`, Clean Architecture folders, axios DB adapters, ZMQ + RPC, epoch zip backups, Express `/tx-start` gate.
**Reason:** PSF already operates `psf-slp-indexer-g2` + `psf-slp-db` in production. Reusing the pattern reduces training cost, Docker layout, and incident playbooks.
**Tradeoff:** Memo does not need SLPs DAG ordering or UTXO graph, so some SLP machinery is absent—but the **process and deployment** model stays parallel.
| Aspect | SLP stack | Memo stack |
|--------|-----------|------------|
| DB port | 5020 | 5021 |
| TX REST port | 5454 | 5455 |
| Parser | `slp-parser` + Lokad ID | `memo-parser` + `0x6d` prefix |
| DB entities | tokens, utxos, addrs | posts, likes, follows, … |
## 2. Separate database service (not embedded LevelDB)
**Decision:** Indexer never opens LevelDB files directly.
**Pros:**
- Aligns with psf-slp-db backup/restore workflow
- Avoids multi-process file lock issues
- Clear boundary for future read replicas or query services
**Cons:**
- HTTP latency on every write
- More moving parts in development (three processes)
**Alternative rejected:** Single Node process with embedded LevelDB—simpler locally but diverges from PSF production norms.
## 3. Scope limited to Go indexer “core social” handlers
**Decision:** v1 implements the handlers in [memo/index/node/obj/op_return/main.go](https://github.com/memocash/index/blob/master/node/obj/op_return/main.go) (`GetHandlers`), not the full [memo-protocol.md](../../memo-protocol.md) table.
**Included:** name, post, reply, like, profile, profile pic, follow/unfollow, topic message, topic follow/unfollow.
**Excluded (v1):**
- Polls, mute, send money
- MIP-0009 token sale actions (`0x6d30``0x6d32`)
- SLP handler present in Gos registry (`slp_tokenHandler`)
- Planned actions (`0x6d08`, `0x6d0b`, …)
**Reason:** Ship a useful social subgraph first; token and poll logic add validation and cross-tx dependencies.
**Risk:** Indexer state diverges from memo.sv for excluded actions until handlers are added.
## 4. Parallel processing within blocks (no DAG sort)
**Decision:** Filter and process Memo txs concurrently within a block using `p-queue`. Default concurrency is 20 for both phases (`FILTER_CONCURRENCY`, `MEMO_TX_CONCURRENCY`). Filter results are reordered to match block tx order before processing starts.
**SLP context:** SEND transactions may spend token outputs created earlier in the same block; DAG sort orders them correctly and writes must stay serial when UTXO state overlaps.
**Memo context:** Social actions do not consume each others UTXOs in a token graph. Most handler writes are keyed by `txid` and are independent across transactions. Parallel `processMemoTx` is safe for typical blocks.
**Soft ordering:** Likes that reference a post in the same block may compute `tip = 0` if the post handler has not finished yet. Replies still persist parent/child links and post bodies regardless of completion order. Profile/follow handlers keyed by address can race if the same signer emits multiple updates in one block (rare).
**Within a single tx:** Multiple Memo `OP_RETURN` outputs in one transaction are still dispatched sequentially inside `processMemoTx`.
**Tradeoff:** Higher throughput vs. HTTP contention on `psf-memo-db`. Lower `MEMO_TX_CONCURRENCY` if the DB service becomes the bottleneck.
**Alternative rejected:** Full serial processing—simple but leaves RPC/DB latency on the table during IBD when blocks contain many unrelated Memo txs.
## 5. Scan all transaction outputs for OP_RETURN
**Decision:** `findMemoOutputs` checks every `vout`, not only `vout[0]`.
**Reason:** Go `node/obj/saver/op_return.go` iterates all outputs; Memo has no protocol rule fixing OP_RETURN to the first output.
**Cost:** Slightly more script parsing per tx during filter and process.
## 6. Custom script pushdata parser (with selective bch-js)
**Decision:** `parseScriptPushDatas` implements Bitcoin push opcode walking in JavaScript; `@psf/bch-js` is used for signer address extraction from inputs.
**Reason:** Unit tests must run without a full node; decompile behavior must be predictable for standard `OP_RETURN` Memo txs.
**Tradeoff:** Non-standard scripts may parse differently than `btcd`—edge cases go to `processErrors`.
## 7. Signer = first decodable input address
**Decision:** Walk `vin` until `Script.getAddressFromScriptSig` succeeds.
**Matches:** Go indexer behavior when building `parse.OpReturn.Addr`.
**Limitation:** Multi-input txs with multiple signers attribute the action to the first resolved address only—consistent with reference indexer, not necessarily with user intent for complex txs.
## 8. Idempotency at transaction level only
**Decision:** `ptxs` store marks completed txs; handlers may write multiple LevelDB keys per tx without a cross-store transaction.
**Go indexer:** Uses richer `db.Save([]db.Object{...})` batches.
**Failure mode:** Crash mid-handler could leave partial writes; re-run skips entire tx if `ptx` was written— or duplicates sub-objects if ptx write failed last.
**Mitigation path:** Write ptx last (current code writes ptx after handlers—good) or add DB batch API.
## 9. Process errors are non-fatal
**Decision:** Invalid pushdata counts or oversize fields log to `processErrors` and return; they do not stop the block.
**Reason:** Chain history contains malformed or experimental Memo txs; one bad tx must not halt IBD.
**Ops note:** Monitor `processErrors` growth for parser bugs vs true invalid chain data.
## 10. Start block 525000
**Decision:** Default `START_BLOCK_HEIGHT=525000` from Go `BeginningOfMemoHeight`.
**Tradeoff:** Earlier Memo experiments exist on-chain but are excluded unless the operator lowers the height and reindexes from genesis of Memo data.
## 11. psf-memo-db: thin controller, no GraphQL
**Decision:** Level routes call LevelDB directly; no Mongoose, no GraphQL.
**Go memo/index** exposes GraphQL with attach resolvers loading from sharded DB—the read path is a separate product layer.
**PSF stack** stops at indexed storage + REST CRUD for writers. Read APIs are the integrators responsibility.
**Future:** A `memo-query.js` adapter (like `slp-query.js`) could encapsulate common reads without pulling in full GraphQL.
## 12. Mempool vs confirmed indexing
**Decision:** TX indexer uses `blockHeight = tip + 1` for unconfirmed txs; block indexer re-processes confirmed blocks.
**Nuance:** A mempool tx may be indexed twice (mempool + block) but `ptx` idempotency prevents duplicate handler effects.
**Edge case:** Tx never confirms—remains indexed with provisional height; may be acceptable for social UX or require GC policy later.
## 13. Docker layout under `production/docker`
**Decision:** Dockerfiles live in `psf-memo-indexer/production/docker/{memo-db,block-indexer,tx-indexer}/`, not repo roots—matching psf-slp-indexer-g2.
**Reason:** Single compose file orchestrates three services; build contexts point at `psf-memo-db` sibling repo and indexer root.
## 14. JavaScript vs extending the Go indexer
**Decision:** New JS implementation rather than operating memocash/index directly.
**Pros for PSF:**
- Same language and patterns as SLP indexer maintenance
- No CGO / Go toolchain requirement for PSF deployers
- Independent release cycle from memocash
**Cons:**
- Two implementations to keep in sync with protocol changes
- Go indexer has sharding, GraphQL, and mature handlers PSF omitted
**When to prefer Go:** Full memo.sv parity, production-scale sharded cluster, SLP-in-Memo token indexing.
**When to prefer PSF JS:** Unified ops with psf-slp-indexer-g2, custom downstream consumers of LevelDB REST.
## 15. Testing strategy
**Decision:** Unit tests with mocks for RPC, DB, and parser fixtures (Go `post_test.go` tx hex)—no mandatory e2e full node in CI.
**Tradeoff:** Integration bugs (RPC format changes, ZMQ topic names) need manual or staged e2e tests.
**Recommended e2e smoke:** Index block ≥ 525000 containing a known Memo post from [memo-protocol.md](../../memo-protocol.md) explorer links; verify `GET /level/post/:txid`.
## Decision log (quick reference)
| Topic | Choice |
|-------|--------|
| Architecture pattern | Clean Architecture |
| Chain interface | RPC + ZMQ |
| Storage | psf-memo-db REST + LevelDB |
| Processes | 2 indexers + 1 DB |
| Protocol breadth | Go `GetHandlers` subset |
| Ordering | Block order, no DAG |
| OP_RETURN location | Any output |
| Auth on DB API | None (v1) |
| Read API | Out of scope (v1) |
## Open questions for future contributors
1. **Handler parity:** Which excluded memo-protocol actions are required next (polls vs token sale)?
2. **Query layer:** Should PSF add `memo-query` REST on psf-memo-db or a separate service?
3. **Reorg handling:** v1 does not unwind reorgs; is depth-0 BCH acceptable for social indexing?
4. **Address indexing:** Should posts be secondary-indexed by `addr` for feed APIs?
5. **Alignment with Go:** Should reply/like tip logic match Gos satoshi aggregation exactly (output script types beyond P2PKH)?
Document updates belong in this `dev-docs/` folder when decisions change.
+87
View File
@@ -0,0 +1,87 @@
# Overview: Why the Memo Indexer Exists
## The problem
The [Memo protocol](https://memo.cash) encodes social actions—posts, replies, likes, follows, profiles, topics—in **Bitcoin Cash transactions** using `OP_RETURN` outputs. Wallets and explorers can read individual transactions from a full node, but useful applications need:
- A **queryable history** of who posted what, when, and in reply to which parent transaction
- **Derived indexes** (follow graph, likes per post, topic membership) without re-scanning the entire chain on every API request
- **Consistent interpretation** of Memo action bytes (`0x6d` prefix + action code) across services
Scanning the blockchain on demand for every user request does not scale. The Memo indexer exists to **materialize** protocol state into local databases so other apps (APIs, analytics, mirrors) can read indexed data quickly.
## What this stack does
Two cooperating Node.js services mirror the proven PSF pattern used for SLP tokens:
```text
BCH full node (RPC + ZMQ)
┌───────────────────────┐ HTTP REST ┌─────────────────┐
│ psf-memo-indexer │ ─────────────────► │ psf-memo-db │
│ (2 processes) │ /level/* │ (LevelDB) │
└───────────────────────┘ └─────────────────┘
│ │
│ ▼
│ Query / backup APIs
Memo OP_RETURN txs
from blocks + mempool
```
**psf-memo-indexer** watches the chain, detects Memo transactions, validates action payloads, and writes structured records.
**psf-memo-db** owns storage. It exposes CRUD over multiple LevelDB instances so the indexer (and future tools) do not open database files directly—avoiding lock contention and allowing backup/restore as a separate concern.
## What v1 indexes
The JavaScript indexer intentionally matches the **core social handlers** registered in the reference Go indexers `op_return` package—not the full Memo protocol spec:
| Indexed in v1 | Not indexed in v1 |
|---------------|-------------------|
| Set name (`0x6d01`) | Polls (`0x6d10``0x6d14`) |
| Post (`0x6d02`) | Mute / unmute |
| Reply (`0x6d03`) | Send money (`0x6d24`) |
| Like (`0x6d04`) | Token sale MIP-0009 actions |
| Set profile (`0x6d05`) | SLP / Bitcom OP_RETURN in Go indexer |
| Follow / unfollow (`0x6d06` / `0x6d07`) | Planned spec actions (`0x6d08`, `0x6d0b`, …) |
| Set profile picture (`0x6d0a`) | |
| Topic message / follow / unfollow (`0x6d0c``0x6d0e`) | |
See [design-decisions-and-tradeoffs.md](./design-decisions-and-tradeoffs.md) for rationale.
## Start height
Indexing begins at block **525000** (`BeginningOfMemoHeight` in the Go reference). Earlier blocks are skipped because Memo activity before that height is negligible for practical deployments.
## Who consumes the output
This repository does **not** ship a GraphQL or public REST API for end users—that exists in the separate [memo/index](https://github.com/memocash/index) Go project. PSFs stack is an **indexer + DB layer** intended for:
- PSF infrastructure that wants Memo data beside SLP indexing patterns
- Custom services that call `psf-memo-db` REST endpoints
- Future query layers (e.g. a slim `memo-query` module) without reimplementing chain scanning
## Operational summary
| Process | Entry point | When it runs |
|---------|-------------|--------------|
| Block indexer | `psf-memo-block-indexer.js` | IBD from `START_BLOCK_HEIGHT` to chain tip; then ZMQ `rawblock` |
| TX indexer | `psf-memo-tx-indexer.js` | After block indexer calls `GET /tx-start`; then ZMQ `rawtx` |
| Database | `psf-memo-db` `index.js` | Must be running before indexers start |
Typical development:
```bash
# Terminal 1
cd psf-memo-db && npm start
# Terminal 2
cd psf-memo-indexer && npm run block-indexer
# Terminal 3
cd psf-memo-indexer && npm run tx-indexer
```
Production uses [production/docker](../production/docker/) with the same three logical services.
+218
View File
@@ -0,0 +1,218 @@
# psf-memo-db Architecture
`psf-memo-db` is the persistence layer for the Memo indexer. It is structurally modeled on [psf-slp-db](https://github.com/Permissionless-Software-Foundation/psf-slp-db): multiple LevelDB instances behind a Koa REST API, with no business logic beyond backup and restore.
## Why a separate service
| Concern | How separation helps |
|---------|----------------------|
| File locking | Only one process opens LevelDB files; indexers use HTTP |
| Parallelism | Multiple indexer workers could share one DB API (future) |
| Backups | Close all DBs, zip, reopen—without stopping indexer logic in the same process |
| Operational familiarity | PSF already runs `psf-slp-db` beside `psf-slp-indexer-g2` |
**Tradeoff:** Every write is an HTTP round trip. Local indexing pays latency versus embedded LevelDB, but gains operational consistency with the SLP stack.
## Server bootstrap
Mirrors `psf-slp-db/bin/server.js`:
```text
index.js
└── bin/server.js
├── Koa + middleware (logger, bodyparser 100mb, CORS, error handler)
├── controllers.initAdapters() → level-db.openDbs()
├── controllers.initUseCases() → no-op start
├── controllers.attachRESTControllers()
└── app.listen(PORT) // default 5021
```
**Intentionally omitted** from psf-slp-db: MongoDB, Passport auth, IPFS/Helia, wallet, usage tracking, `/slp` query routes. `noMongo: true` and `useIpfs: false` in config.
## LevelDB instances
Opened in `src/adapters/level-db.js`:
```javascript
leveldb/current/{name} // valueEncoding: 'json'
```
| Instance | Cache hint | Indexer writes |
|----------|------------|----------------|
| `status` | 64 MB | Block height sync state |
| `posts` | 512 MB | Post and reply bodies |
| `postParents` | 64 MB | Reply → parent link |
| `postChildren` | 64 MB | Parent → children |
| `likes` | 64 MB | Like events |
| `names` | 64 MB | Display names |
| `profiles` | 64 MB | Profile text |
| `profilePics` | 64 MB | Avatar URLs |
| `follows` | 64 MB | Follow graph edges |
| `rooms` | 64 MB | Topic posts and follows |
| `processErrors` | 64 MB | Skipped / invalid txs |
| `ptxs` | 64 MB | Processed tx markers |
Posts receive a larger cache because they are the highest-volume social object.
## REST API shape
All routes are under `/level` with a consistent CRUD pattern generated from `ENTITY_CONFIG` in `crud-handlers.js`:
| Method | Path pattern | Body (create) |
|--------|--------------|---------------|
| `POST` | `/level/{entity}` | `{ <idField>, <dataField> }` |
| `GET` | `/level/{entity}/:key` | — |
| `PUT` | `/level/{entity}/:key` | `{ <dataField> }` |
| `DELETE` | `/level/{entity}/:key` | — |
### Entity route names
| Route | idField | dataField | Example key |
|-------|---------|-----------|-------------|
| `post` | `txid` | `postData` | transaction id |
| `postparent` | `txid` | `parentData` | child txid |
| `postchild` | `key` | `childData` | `parentTxid:childTxid` |
| `like` | `txid` | `likeData` | like txid |
| `name` | `addr` | `nameData` | cash address |
| `profile` | `addr` | `profileData` | cash address |
| `profilepic` | `addr` | `profilePicData` | cash address |
| `follow` | `key` | `followData` | `follower:followeePkHash` |
| `room` | `key` | `roomData` | composite |
| `processerror` | `txid` | `errorData` | txid |
| `ptx` | `txid` | `ptxData` | txid |
### Read routes (query use cases)
| Method | Path | Query params |
|--------|------|----------------|
| `GET` | `/profile/recent` | `limit` (default 100, max 100), `offset` (default 0) |
| `GET` | `/posts/recent` | `limit` (default 100, max 100), `offset` (default 0) |
Returns profiles or posts sorted by **block height** (newest first), using the `blockHeight` field stored on each entity document at indexing time. Tie-breaker: `seen` timestamp descending.
The `seen` field is **Unix epoch milliseconds** from the block header time (`block.time * 1000` at indexing time).
Response shape (`/profile/recent`):
```json
{
"profiles": [
{
"addr": "bitcoincash:q...",
"text": "...",
"txid": "...",
"seen": 1500000000000,
"blockHeight": 600000
}
],
"pagination": { "limit": 100, "offset": 0, "total": 42, "hasMore": false }
}
```
Response shape (`/posts/recent`):
```json
{
"posts": [
{
"txid": "...",
"addr": "bitcoincash:q...",
"text": "...",
"seen": 1500000000000,
"blockHeight": 600000
}
],
"pagination": { "limit": 100, "offset": 0, "total": 42, "hasMore": false }
}
```
Implementation: `profile-query` / `post-query` adapter (LevelDB scan) → `list-recent-profiles` / `list-recent-posts` use case → REST controller.
**Tradeoff:** Full scan of `profiles` on each request; suitable for moderate corpus sizes. A height-indexed store would be needed for very large archives.
### Status (special case)
Matches SLP status semantics:
- `GET /level/status/:statusKey`
- `POST /level/status` with `{ statusKey, statusData }`
- `PUT /level/status` with `{ statusData }` — key is always `status`
- `DELETE /level/status/:statusKey`
Indexer expects `statusKey = status` containing:
```json
{
"startBlockHeight": 524999,
"syncedBlockHeight": 800000,
"chainBlockHeight": 800001
}
```
### Backup and restore
- `POST /level/backup` — body `{ height, epoch }` → zip `leveldb/current` to `leveldb/zips/memo-indexer-{height}.zip`
- `POST /level/restore` — body `{ height }` → unzip matching archive and `process.exit(0)`
Implemented in `src/adapters/db-backup.js`.
### Health
`GET /health``{ status: 'ok' }` for load balancers and compose checks.
## Indexer adapter mapping
The indexer uses `src/adapters/entity-db.js`:
```javascript
createEntityDb('post', 'txid', 'postData')
// → POST http://localhost:5021/level/post
```
`status-db.js` uses dedicated status endpoints rather than the generic factory.
## Data modeling notes
**Not normalized like SQL.** LevelDB stores are document keyed for fast lookup by txid or address, similar to the Go `db/item/memo` objects but without sharding.
**Denormalized block height.** Every entity written by the indexer includes a `blockHeight` field (the block in which the memo transaction was confirmed, or `tip + 1` for unconfirmed txs). This avoids ptx lookups when serving `/recent` query routes. The `ptxs` store remains for idempotency only.
Common fields on indexed documents:
| Entity | Key | Stored fields (includes) |
|--------|-----|--------------------------|
| post | txid | `addr`, `text`, `seen`, `blockHeight` |
| profile | addr | `text`, `txid`, `seen`, `addr`, `blockHeight` |
| name | addr | `name`, `txid`, `seen`, `addr`, `blockHeight` |
| profilePic | addr | `url`, `txid`, `seen`, `addr`, `blockHeight` |
| like | txid | `addr`, `postTxid`, `seen`, `tip`, `blockHeight` |
| follow | composite key | `followerAddr`, `followeePkHash`, `unfollow`, `txid`, `seen`, `blockHeight` |
| postParent / postChild | txid / `parentTxid:childTxid` | `parentTxid`, `childTxid`, `blockHeight` |
| room | composite key | `room`, `txid`, `seen`, `type`, `blockHeight` (+ `addr` for follows) |
| processError | txid | `error`, `ts`, `blockHeight` |
**No secondary indexes in v1.** Queries like “all posts by address” may require scanning or a future `memo-query` adapter—out of scope for the indexer write path.
**JSON values** keep debugging simple; binary serialization would save space but break parity with psf-slp-db tooling.
## Configuration
| Variable | Default | Purpose |
|----------|---------|---------|
| `PORT` | `5021` | REST listen (SLP uses 5020) |
| `SVC_ENV` | `development` | Config profile |
| `BACKUP_QTY` | `3` | Retained zip backups |
| `EXIT_ON_MISSING_BACKUP` | `false` | Fail restore if zip missing |
## Testing
Unit tests cover:
- `bin/server.js` startup with stubbed controllers
- Level controller status and post CRUD with mocked LevelDB
Run: `npm test` in the `psf-memo-db` directory.
## Production
Docker build context is the `psf-memo-db` repo root; Dockerfile lives in [psf-memo-indexer/production/docker/memo-db/](../production/docker/memo-db/). See [architecture.md](./architecture.md#deployment-topology).
@@ -0,0 +1,196 @@
# Theory of Operation
This document walks through how the indexer behaves at runtime—from cold start through steady-state chain following—including the Memo-specific parsing rules.
## Prerequisites
1. **BCH full node** with RPC and ZMQ enabled (`rawtx`, `rawblock` on the configured port, default `28332`).
2. **psf-memo-db** listening (default `http://localhost:5021`).
3. Environment variables in `.env` or Docker-mounted `.env` files (see `.env-example`).
## Phase 1: Initialization
Both indexer processes:
1. Construct `Adapters`, `UseCases`, and `Controllers`.
2. Block indexer enables keyboard listener (`q` stops after current block).
3. TX indexer starts Express on `TX_REST_API_PORT` and waits for `/tx-start`.
**Status bootstrap** (`status-db.js`):
- `GET /level/status/status` from psf-memo-db.
- If missing (fresh DB), create:
```json
{
"startBlockHeight": START_BLOCK_HEIGHT - 1,
"syncedBlockHeight": START_BLOCK_HEIGHT - 1,
"chainBlockHeight": <current RPC block count>
}
```
If `EXIT_ON_MISSING_BACKUP=true` and status is missing, the process exits instead of reinitializing—same safety valve as the SLP indexer when a backup is expected but absent.
## Phase 2: Initial Block Download (IBD)
Only the **block indexer** runs IBD.
```text
nextHeight = syncedBlockHeight + 1
tip = RPC getblockcount
while nextHeight <= tip:
processBlock(nextHeight)
updateIndexedBlockHeight(nextHeight)
optionally backup every 1000 blocks
nextHeight++
```
### Per-block pipeline (`index-blocks.processBlock`)
```mermaid
sequenceDiagram
participant BI as Block_indexer
participant RPC as Full_node_RPC
participant FB as filter_block
participant TX as transaction_adapter
participant H as action_handlers
participant DB as psf_memo_db
BI->>RPC: getblockhash(height)
BI->>RPC: getblock(hash)
BI->>FB: filterMemoTxs(txids[])
loop each candidate txid
FB->>TX: isMemoTx(txid)
end
loop each memo txid (parallel, up to MEMO_TX_CONCURRENCY)
BI->>BI: processMemoTx(txid)
BI->>DB: reads/writes via handlers
end
```
**Step 1 — Fetch block:** `getblock` returns the list of transaction ids in the block (verbosity 1).
**Step 2 — Filter:** `filterMemoTxs` runs up to `FILTER_CONCURRENCY` (default 20) concurrent `isMemoTx` checks (`p-queue`). Each check loads the transaction (with in-memory cache in `transaction.js`) and scans **all outputs** for a Memo `OP_RETURN`. Results are returned in **block tx order** (parallel checks use a set, then the original block list is filtered).
**Why scan all outputs?** Unlike SLP, which conventionally places token data in `vout[0]`, Memo actions may appear in any outputs `scriptPubKey`. The Go reference iterates every `TxOut`.
**Step 3 — Process:** `processMemoTxs` runs up to `MEMO_TX_CONCURRENCY` (default 20) concurrent `processMemoTx` calls via `p-queue`. There is no DAG sort—Memo social actions do not form token-like dependency chains within a block. Multiple OP_RETURN outputs in a **single** transaction are still handled sequentially inside `processMemoTx`.
## Phase 3: Processing a single Memo transaction
### Idempotency
Before heavy work, the indexer checks `GET /level/ptx/{txid}`. If present, the tx is skipped. After successful handling, it `POST`s a ptx record `{ processedAt, blockHeight }`.
- **`seen`** on indexed entities (profiles, posts, likes, etc.) is the on-chain block header time in **Unix epoch milliseconds** (`block.time * 1000` from the BCH full node).
- **`processedAt`** on ptx records is indexer wall-clock time in milliseconds (`Date.now()` when the tx was processed).
**Nuance:** If a handler partially fails mid-tx, v1 does not roll back prior writes in that tx. The Go indexer uses richer DB transactions; PSF v1 relies on idempotency only at tx boundaries. See tradeoffs doc.
### Signer address
Memo actions are attributed to the **first P2PKH input** that yields an address from the unlocking script (`getSignerAddress` in `memo-parser.js`, backed by `@psf/bch-js` where possible).
If no address is found:
- Log `processErrors` with reason.
- Skip action handlers (no anonymous posts in v1).
This mirrors the Go `SetLockHash` loop over inputs.
### OP_RETURN decoding
`parseScriptPushDatas` walks Bitcoin script bytes (push opcodes `0x01``0x4e`, `OP_RETURN 0x6a`, etc.) and collects pushdata buffers.
A Memo action is recognized when **any** pushdata begins with:
```text
0x6d <action_byte>
```
`decodeMemoOpReturn` returns `{ action, prefix, pushDatas }` where `action` is a string key (`post`, `reply`, `like`, …) mapped in `memo-codes.js`.
### Handler dispatch
`dispatchMemoAction` routes to `src/use-cases/action-types/*.js`. Each handler:
1. Validates pushdata count and field sizes (against `MAX_POST_SIZE` etc. from Go `memo.go`, typically 65000 bytes in modern limits).
2. On validation failure: write `processErrors`, return without throwing (invalid Memo txs do not halt the block).
3. On success: `POST` / `PUT` to the appropriate `/level/{entity}` route.
### Multi-output transactions
If one transaction contains multiple Memo `OP_RETURN` outputs, each decoded output dispatches separately in a loop. Rare in practice but supported.
## Action-specific behavior (v1)
| Action | Pushdata layout | Side effects beyond primary store |
|--------|-----------------|-----------------------------------|
| `setName` | prefix + UTF-8 name | `names` by address |
| `post` | prefix + message | `posts` by txid; skip create if post exists |
| `reply` | prefix + 32-byte parent hash + message | `postParents`, `postChildren`, then post body |
| `like` | prefix + 32-byte post hash | `likes`; optional `tip` satoshis to post author outputs |
| `setProfile` | prefix + text | `profiles` |
| `follow` / `unfollow` | prefix + 20-byte pk hash | `follows` keyed by follower and followee |
| `setProfilePic` | prefix + URL | `profilePics` |
| `topicMessage` | prefix + topic + message | post + `rooms` index |
| `topicFollow` / `topicUnfollow` | prefix + topic name | `rooms` follow record |
Like-tip logic: if the liked post exists and the liker is not the author, outputs paying the authors address are summed (values converted from BCH decimal to satoshis) and stored on the like record.
## Phase 4: IBD completion and TX indexer handoff
When `nextBlockHeight > tip`:
1. Block indexer connects ZMQ.
2. `GET /tx-start` on TX indexer.
3. Enters ZMQ block loop (500 ms sleep, `rawblock` queue).
TX indexer:
1. Already listening on Express.
2. After `/tx-start`, connects ZMQ and processes `rawtx`.
3. Maintains `seenTxs` Set + FIFO queue capped at `seenTxMax` to drop duplicate ZMQ notifications cheaply before hitting `ptxDb`.
4. Calls the same `processMemoTx` with `blockHeight = chain tip + 1` for unconfirmed txs.
## Phase 5: Steady state
| Event source | Process | Behavior |
|--------------|---------|----------|
| New block (ZMQ) | Block indexer | Resolve height from block header, update `status`, `processBlock` |
| New mempool tx (ZMQ) | TX indexer | Dedupe → `processMemoTx` |
| Every 1000 blocks | Block indexer | `POST /level/backup` with `{ height, epoch: 1000 }` |
### Backups
`psf-memo-db` closes LevelDB files, zips `leveldb/current` to `leveldb/zips/memo-indexer-{height}.zip`, reopens DBs. Old zips are pruned per `BACKUP_QTY`.
Restore (`POST /level/restore`) unzips and **exits the process**—expect process manager restart, same as SLP.
## Memo vs SLP detection (operational difference)
| | SLP indexer | Memo indexer |
|---|-------------|--------------|
| Detection | Lokad `534c5000` + `slp-parser` | Prefix `6d` + action byte |
| Output index | Assumes `vout[0]` for token data | Any vout |
| Ordering | DAG sort within block | Sequential tx order |
| State model | Token UTXOs, balances | Social graph documents |
## Failure and retry behavior
- **RPC:** `RetryQueue` and `p-retry` (5 attempts) on block filter operations.
- **DB HTTP:** Errors bubble up; block processing logs and throws on `processMemoTx` failure.
- **Invalid Memo payload:** Logged to `processErrors`; block continues.
- **Keyboard `q`:** Block indexer exits after finishing current block.
## Configuration reference
| Variable | Default | Effect |
|----------|---------|--------|
| `PSF_MEMO_DB_URL` | `http://localhost:5021` | All DB adapters |
| `START_BLOCK_HEIGHT` | `525000` | Genesis for indexer state |
| `RPC_*` / `ZMQ_PORT` | see `.env-example` | Full node |
| `TX_REST_API_PORT` | `5455` | TX control (SLP uses 5454) |
| `seenTxMax` | `100000` | Mempool dedupe set size |
| `txCacheMax` | `100000` | RPC tx cache in transaction adapter |
+10131
View File
File diff suppressed because it is too large Load Diff
+32
View File
@@ -0,0 +1,32 @@
{
"name": "psf-memo-indexer",
"version": "1.0.0",
"description": "Indexes Memo protocol transactions on Bitcoin Cash.",
"type": "module",
"scripts": {
"block-indexer": "node --max-old-space-size=8192 psf-memo-block-indexer.js",
"tx-indexer": "node --max-old-space-size=4096 psf-memo-tx-indexer.js",
"test": "c8 --reporter=text mocha --exit --recursive test/unit/",
"lint": "standard --env mocha --fix"
},
"author": "Chris Troutner",
"license": "GPLV3",
"dependencies": {
"@chris.troutner/retry-queue": "1.0.11",
"@psf/bch-js": "7.1.11",
"@psf/bitcoincash-zmq-decoder": "0.1.5",
"axios": "1.12.2",
"dotenv": "17.2.3",
"express": "5.1.0",
"p-queue": "8.1.1",
"p-retry": "6.2.1",
"zeromq": "6.5.0"
},
"devDependencies": {
"c8": "10.1.3",
"chai": "6.2.0",
"mocha": "11.7.4",
"sinon": "21.0.0",
"standard": "17.1.2"
}
}
@@ -0,0 +1,19 @@
# Environment variables for psf-memo-block-indexer
# Copy this file to .env and modify as needed
PSF_MEMO_DB_URL=http://172.17.0.1:5021
RPC_IP=172.17.0.1
RPC_PORT=8332
ZMQ_PORT=28332
RPC_USER=bitcoin
RPC_PASS=password
TX_REST_API_PORT=5455
TX_REST_API_IP=172.17.0.1
START_BLOCK_HEIGHT=525000
EXIT_ON_MISSING_BACKUP=false
FILTER_CONCURRENCY=20
MEMO_TX_CONCURRENCY=20
DEBUG_LEVEL=0
@@ -0,0 +1,50 @@
# Create a Dockerized block indexer
#
#IMAGE BUILD COMMANDS
FROM ubuntu:24.04
MAINTAINER Chris Troutner <chris.troutner@gmail.com>
#Update the OS and install any OS packages needed.
RUN apt-get update
RUN apt-get install -y sudo git curl nano gnupg wget zip unzip python3
#Install Node and NPM
RUN curl -sL https://deb.nodesource.com/setup_22.x -o nodesource_setup.sh
RUN bash nodesource_setup.sh
RUN apt-get install -y nodejs build-essential
#Create the user 'safeuser' and add them to the sudo group.
RUN useradd -ms /bin/bash safeuser
RUN adduser safeuser sudo
#Set password to 'password' change value below if you want a different password
RUN echo safeuser:password | chpasswd
#Set the working directory to be the home directory
WORKDIR /home/safeuser
#Setup NPM for non-root global install
RUN mkdir /home/safeuser/.npm-global
RUN chown -R safeuser .npm-global
RUN echo "export PATH=~/.npm-global/bin:$PATH" >> /home/safeuser/.profile
RUN runuser -l safeuser -c "npm config set prefix '~/.npm-global'"
# Clone the psf-memo-indexer repository
WORKDIR /home/safeuser
RUN git clone https://github.com/Permissionless-Software-Foundation/psf-memo-indexer
# Switch to the desired branch. `master` is usually stable,
# and `stage` has the most up-to-date changes.
WORKDIR /home/safeuser/psf-memo-indexer
# For development: switch to unstable branch
#RUN git checkout <branch-name>
# Install dependencies
RUN npm install
# Start the application.
# start-block-indexer.sh is mounted at runtime via docker-compose.
VOLUME start-block-indexer.sh
CMD ["./start-block-indexer.sh"]
@@ -0,0 +1,3 @@
#!/bin/bash
node --max-old-space-size=8192 psf-memo-block-indexer.js
@@ -0,0 +1,61 @@
# Start the service with: docker-compose up -d
services:
memo-db:
build: ./memo-db
container_name: memo-db
logging:
driver: json-file
options:
max-size: 10m
max-file: '10'
ports:
- '5021:5021'
volumes:
- ../data/leveldb:/home/safeuser/psf-memo-db/leveldb
- ./memo-db/start-memo-db.sh:/home/safeuser/psf-memo-db/start-memo-db.sh
- ./memo-db/.env:/home/safeuser/psf-memo-db/.env
restart: always
tx-indexer:
build: ./tx-indexer
image: memo-tx-indexer
container_name: memo-tx-indexer
logging:
driver: json-file
options:
max-size: 10m
max-file: '10'
ports:
- '5455:5455'
volumes:
- ./tx-indexer/start-tx-indexer.sh:/home/safeuser/psf-memo-indexer/start-tx-indexer.sh
- ./tx-indexer/.env:/home/safeuser/psf-memo-indexer/.env
restart: always
block-indexer:
build: ./block-indexer
image: memo-block-indexer
container_name: memo-block-indexer
logging:
driver: json-file
options:
max-size: 10m
max-file: '10'
volumes:
- ./block-indexer/start-block-indexer.sh:/home/safeuser/psf-memo-indexer/start-block-indexer.sh
- ./block-indexer/.env:/home/safeuser/psf-memo-indexer/.env
restart: always
memo-client:
build: ./memo-client
image: memo-client
container_name: memo-client
logging:
driver: json-file
options:
max-size: 10m
max-file: '10'
ports:
- '3000:80'
restart: always
@@ -0,0 +1,9 @@
# psf-memo-db REST API base URL (no trailing slash).
# Baked into the SPA when the memo-client image is built.
# Example: https://api.mydomain.com
#
# Usage:
# cp .env-example .env
# # edit REACT_APP_MEMO_DB_URL
# docker compose build memo-client
REACT_APP_MEMO_DB_URL=http://localhost:5021
@@ -0,0 +1,48 @@
# Create a Dockerized front-end SPA served by nginx
#
#IMAGE BUILD COMMANDS
FROM ubuntu:24.04 as builder
MAINTAINER Chris Troutner <chris.troutner@gmail.com>
#Update the OS and install any OS packages needed.
RUN apt-get update
RUN apt-get install -y sudo git curl nano gnupg wget
#Install Node and NPM
RUN curl -sL https://deb.nodesource.com/setup_22.x -o nodesource_setup.sh
RUN bash nodesource_setup.sh
RUN apt-get install -y nodejs build-essential
#Set the working directory to be the home directory
WORKDIR /home/safeuser
# Clone the psf-memo-client repository
RUN git clone https://github.com/Permissionless-Software-Foundation/psf-memo-client
# Switch to the desired branch. `master` is usually stable,
# and `stage` / `unstable` have the most up-to-date changes.
WORKDIR /home/safeuser/psf-memo-client
# For development: switch to unstable branch
#RUN git checkout unstable
# Install dependencies
RUN npm install
# Create React App reads REACT_APP_* from this .env at build time.
# Edit production/docker/memo-client/.env before building.
COPY .env .env
# Build the site
RUN npm run build
# Load the NGINX image.
FROM nginx
EXPOSE 80
# Copy the files built in the first container to the new NGINX container.
COPY --from=builder /home/safeuser/psf-memo-client/build /usr/share/nginx/html
# Copy the NGINX configuration file.
COPY default.conf /etc/nginx/conf.d/default.conf
@@ -0,0 +1,22 @@
server {
listen 80;
listen [::]:80;
server_name localhost;
#access_log /var/log/nginx/host.access.log main;
location / {
root /usr/share/nginx/html;
index index.html index.htm;
try_files $uri $uri/ /index.html;
}
#error_page 404 /404.html;
# redirect server error pages to the static page /50x.html
#
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root /usr/share/nginx/html;
}
}
@@ -0,0 +1,4 @@
PORT=5021
SVC_ENV=prod
BACKUP_QTY=3
EXIT_ON_MISSING_BACKUP=false
@@ -0,0 +1,56 @@
# Create a Dockerized API server
#
#IMAGE BUILD COMMANDS
FROM ubuntu:24.04
MAINTAINER Chris Troutner <chris.troutner@gmail.com>
#Update the OS and install any OS packages needed.
RUN apt-get update
RUN apt-get install -y sudo git curl nano gnupg wget zip unzip python3
#Install Node and NPM
RUN curl -sL https://deb.nodesource.com/setup_22.x -o nodesource_setup.sh
RUN bash nodesource_setup.sh
RUN apt-get install -y nodejs build-essential
#Create the user 'safeuser' and add them to the sudo group.
RUN useradd -ms /bin/bash safeuser
RUN adduser safeuser sudo
#Set password to 'password' change value below if you want a different password
RUN echo safeuser:password | chpasswd
#Set the working directory to be the home directory
WORKDIR /home/safeuser
#Setup NPM for non-root global install
RUN mkdir /home/safeuser/.npm-global
RUN chown -R safeuser .npm-global
RUN echo "export PATH=~/.npm-global/bin:$PATH" >> /home/safeuser/.profile
RUN runuser -l safeuser -c "npm config set prefix '~/.npm-global'"
# Clone the psf-memo-db repository
WORKDIR /home/safeuser
RUN git clone https://github.com/Permissionless-Software-Foundation/psf-memo-db
# Switch to the desired branch. `master` is usually stable,
# and `stage` has the most up-to-date changes.
WORKDIR /home/safeuser/psf-memo-db
# For development: switch to unstable branch
#RUN git checkout <branch-name>
# Install dependencies
RUN npm install
# Generate the API docs
RUN npm run docs
# Expose the port the API will be served on.
EXPOSE 5021
# Start the application.
# start-memo-db.sh is mounted at runtime via docker-compose.
VOLUME start-memo-db.sh
CMD ["./start-memo-db.sh"]
@@ -0,0 +1,3 @@
#!/bin/bash
node index.js
@@ -0,0 +1,15 @@
# Environment variables for psf-memo-tx-indexer
# Copy this file to .env and modify as needed
PSF_MEMO_DB_URL=http://172.17.0.1:5021
RPC_IP=172.17.0.1
RPC_PORT=8332
ZMQ_PORT=28332
RPC_USER=bitcoin
RPC_PASS=password
TX_REST_API_PORT=5455
TX_REST_API_IP=172.17.0.1
SEEN_TX_MAX=100000
@@ -0,0 +1,53 @@
# Create a Dockerized tx indexer
#
#IMAGE BUILD COMMANDS
FROM ubuntu:24.04
MAINTAINER Chris Troutner <chris.troutner@gmail.com>
#Update the OS and install any OS packages needed.
RUN apt-get update
RUN apt-get install -y sudo git curl nano gnupg wget zip unzip python3
#Install Node and NPM
RUN curl -sL https://deb.nodesource.com/setup_22.x -o nodesource_setup.sh
RUN bash nodesource_setup.sh
RUN apt-get install -y nodejs build-essential
#Create the user 'safeuser' and add them to the sudo group.
RUN useradd -ms /bin/bash safeuser
RUN adduser safeuser sudo
#Set password to 'password' change value below if you want a different password
RUN echo safeuser:password | chpasswd
#Set the working directory to be the home directory
WORKDIR /home/safeuser
#Setup NPM for non-root global install
RUN mkdir /home/safeuser/.npm-global
RUN chown -R safeuser .npm-global
RUN echo "export PATH=~/.npm-global/bin:$PATH" >> /home/safeuser/.profile
RUN runuser -l safeuser -c "npm config set prefix '~/.npm-global'"
# Clone the psf-memo-indexer repository
WORKDIR /home/safeuser
RUN git clone https://github.com/Permissionless-Software-Foundation/psf-memo-indexer
# Switch to the desired branch. `master` is usually stable,
# and `stage` has the most up-to-date changes.
WORKDIR /home/safeuser/psf-memo-indexer
# For development: switch to unstable branch
#RUN git checkout <branch-name>
# Install dependencies
RUN npm install
# Expose the port the API will be served on.
EXPOSE 5455
# Start the application.
# start-tx-indexer.sh is mounted at runtime via docker-compose.
VOLUME start-tx-indexer.sh
CMD ["./start-tx-indexer.sh"]
@@ -0,0 +1,3 @@
#!/bin/bash
node --max-old-space-size=4096 psf-memo-tx-indexer.js
+105
View File
@@ -0,0 +1,105 @@
/*
Entry point for the Memo block indexer.
*/
import RetryQueue from '@chris.troutner/retry-queue'
import 'dotenv/config'
import Adapters from './src/adapters/adapters-index.js'
import UseCases from './src/use-cases/use-cases-index.js'
import Controllers from './src/controllers/controllers-index.js'
const EPOCH = 1000
async function start () {
try {
const adapters = new Adapters()
await adapters.initAdapters()
const useCases = new UseCases({ adapters })
await useCases.initUseCases()
const controllers = new Controllers({ useCases, adapters })
await controllers.initControllers()
const queue = new RetryQueue()
console.log('Starting Memo block indexer...')
const status = await useCases.state.getStatus()
console.log('Indexer State:', status)
let nextBlockHeight = status.syncedBlockHeight + 1
let biggestBlockHeight = await queue.addToQueue(adapters.rpc.getBlockCount, {})
if (nextBlockHeight <= biggestBlockHeight) {
do {
const blockStart = new Date()
await useCases.indexBlocks.processBlock(nextBlockHeight)
const blockProcessTime = new Date().getTime() - blockStart.getTime()
console.log(`Block ${nextBlockHeight} processed in ${blockProcessTime / 1000}s`)
nextBlockHeight = await useCases.state.updateIndexedBlockHeight({
lastIndexedBlockHeight: nextBlockHeight
})
if (controllers.keyboard.stopStatus()) {
console.log(`Stopped at block ${nextBlockHeight - 1}`)
process.exit(1)
}
if (nextBlockHeight % EPOCH === 0) {
console.log(`Creating DB backup at block ${nextBlockHeight}`)
await adapters.dbCtrl.backupDb(nextBlockHeight, EPOCH)
}
biggestBlockHeight = await queue.addToQueue(adapters.rpc.getBlockCount, {})
} while (nextBlockHeight <= biggestBlockHeight)
} else {
console.log(`Already at tip (block ${status.syncedBlockHeight}).`)
}
console.log(`\nIBD complete. Last block: ${nextBlockHeight - 1}`)
await adapters.zmq.connect()
console.log('Connected to ZMQ.')
await adapters.txIndexerAdapter.startTxIndexer()
console.log('TX indexer started.')
let loopCnt = 0
const liveStatus = status
do {
let blockHeight = await queue.addToQueue(adapters.rpc.getBlockCount, {})
const block = adapters.zmq.getBlock()
if (block) {
const blockHeader = await queue.addToQueue(
adapters.rpc.getBlockHeader,
block.hash
)
blockHeight = blockHeader.height
liveStatus.syncedBlockHeight = blockHeight
liveStatus.chainBlockHeight = blockHeight
await adapters.statusDb.updateStatus(liveStatus)
await useCases.indexBlocks.processBlock(blockHeight)
}
loopCnt++
if (loopCnt > 100) {
loopCnt = 0
console.log(`ZMQ alive. Block height: ${blockHeight}`)
}
await useCases.utils.sleep(500)
} while (1)
} catch (err) {
console.error('Error in psf-memo-block-indexer:', err)
process.exit(1)
}
}
start()
+69
View File
@@ -0,0 +1,69 @@
/*
Entry point for the Memo TX indexer (mempool).
*/
import RetryQueue from '@chris.troutner/retry-queue'
import 'dotenv/config'
import Adapters from './src/adapters/adapters-index.js'
import UseCases from './src/use-cases/use-cases-index.js'
import Controllers from './src/controllers/controllers-index.js'
import config from './config/index.js'
async function start () {
try {
const queue = new RetryQueue()
const adapters = new Adapters()
const useCases = new UseCases({ adapters })
const controllers = new Controllers({ useCases, adapters })
await controllers.initControllers()
await controllers.startTxRESTController()
console.log('Starting Memo TX indexer...')
let runTxIndexer = false
const seenTxs = new Set()
const seenTxQueue = []
do {
await useCases.utils.sleep(2000)
runTxIndexer = controllers.txRESTController.runTxIndexing
if (runTxIndexer) {
console.log('TX Indexer triggered from REST API!')
break
}
} while (1)
await adapters.zmq.connect()
console.log('Connected to ZMQ.')
do {
const blockHeight = await queue.addToQueue(adapters.rpc.getBlockCount, {})
const tx = adapters.zmq.getTx()
if (tx) {
if (seenTxs.has(tx)) continue
seenTxs.add(tx)
seenTxQueue.push(tx)
if (seenTxQueue.length > config.seenTxMax) {
const oldTxid = seenTxQueue.shift()
seenTxs.delete(oldTxid)
}
try {
await useCases.indexBlocks.processMemoTx(tx, blockHeight + 1, Date.now())
} catch (err) {
console.error(`Error indexing mempool tx ${tx}:`, err.message)
}
}
await useCases.utils.sleep(100)
} while (1)
} catch (err) {
console.error('Error in psf-memo-tx-indexer:', err)
process.exit(1)
}
}
start()
@@ -0,0 +1,43 @@
/*
Top-level adapters index.
*/
import StatusDb from './status-db.js'
import RPC from './rpc.js'
import Transaction from './transaction.js'
import ZMQ from './zmq.js'
import TxIndexerAdapter from './tx-indexer.js'
import DbCtrl from './backup-db.js'
import { createEntityDb } from './entity-db.js'
class Adapters {
constructor (localConfig = {}) {
this.statusDb = new StatusDb()
this.rpc = new RPC()
this.transaction = new Transaction(localConfig)
this.zmq = new ZMQ()
this.txIndexerAdapter = new TxIndexerAdapter()
this.dbCtrl = new DbCtrl()
this.postDb = createEntityDb('post', 'txid', 'postData')
this.postParentDb = createEntityDb('postparent', 'txid', 'parentData')
this.postChildDb = createEntityDb('postchild', 'key', 'childData')
this.likeDb = createEntityDb('like', 'txid', 'likeData')
this.nameDb = createEntityDb('name', 'addr', 'nameData')
this.profileDb = createEntityDb('profile', 'addr', 'profileData')
this.profilePicDb = createEntityDb('profilepic', 'addr', 'profilePicData')
this.followDb = createEntityDb('follow', 'key', 'followData')
this.roomDb = createEntityDb('room', 'key', 'roomData')
this.processErrorDb = createEntityDb('processerror', 'txid', 'errorData')
this.ptxDb = createEntityDb('ptx', 'txid', 'ptxData')
this.initAdapters = this.initAdapters.bind(this)
}
async initAdapters () {
console.log('Adapter libraries initialized.')
return true
}
}
export default Adapters
@@ -0,0 +1,21 @@
/*
Database backup/restore via psf-memo-db REST API.
*/
import axios from 'axios'
import config from '../../config/index.js'
class DbCtrl {
constructor () {
this.axios = axios
this.config = config
this.backupDb = this.backupDb.bind(this)
}
async backupDb (height, epoch) {
await this.axios.post(`${this.config.psfMemoDbUrl}/level/backup`, { height, epoch })
return true
}
}
export default DbCtrl
@@ -0,0 +1,30 @@
/*
Generic axios client for psf-memo-db /level CRUD routes.
*/
import axios from 'axios'
import config from '../../config/index.js'
export function createEntityDb (route, idField, dataField) {
return {
async get (key) {
const response = await axios.get(`${config.psfMemoDbUrl}/level/${route}/${key}`)
return response.data
},
async create (key, data) {
const body = { [idField]: key, [dataField]: data }
const response = await axios.post(`${config.psfMemoDbUrl}/level/${route}`, body)
return response.data
},
async update (key, data) {
const response = await axios.put(`${config.psfMemoDbUrl}/level/${route}/${key}`, {
[dataField]: data
})
return response.data
},
async delete (key) {
const response = await axios.delete(`${config.psfMemoDbUrl}/level/${route}/${key}`)
return response.data
}
}
}
+85
View File
@@ -0,0 +1,85 @@
/*
JSON RPC adapter for BCH full node.
*/
import axios from 'axios'
import config from '../../config/index.js'
class RPC {
constructor () {
this.axios = axios
this.config = config
this.getAxiosOptions = this.getAxiosOptions.bind(this)
this.getBlockCount = this.getBlockCount.bind(this)
this.getBlockHeader = this.getBlockHeader.bind(this)
this.getBlock = this.getBlock.bind(this)
this.getBlockHash = this.getBlockHash.bind(this)
this.getRawTransaction = this.getRawTransaction.bind(this)
}
getAxiosOptions () {
return {
method: 'post',
baseURL: `http://${this.config.rpcIp}:${this.config.rpcPort}/`,
timeout: 15000,
auth: {
username: this.config.rpcUser,
password: this.config.rpcPass
},
data: { jsonrpc: '1.0' }
}
}
async getBlockCount () {
const options = this.getAxiosOptions()
options.data.id = 'getblockcount'
options.data.method = 'getblockcount'
options.data.params = []
const response = await this.axios.request(options)
return response.data.result
}
async getBlockHeader (hash, verbose = true) {
if (!hash) throw new Error('Block hash must be provided')
const options = this.getAxiosOptions()
options.data.id = 'getblockheader'
options.data.method = 'getblockheader'
options.data.params = [hash, verbose]
const response = await this.axios.request(options)
return response.data.result
}
async getBlock (inObj = {}) {
const { hash, verbose = true } = inObj
if (!hash) throw new Error('Block hash must be provided')
const options = this.getAxiosOptions()
options.data.id = 'getblock'
options.data.method = 'getblock'
options.data.params = [hash, verbose]
const response = await this.axios.request(options)
return response.data.result
}
async getBlockHash (inObj = {}) {
const { height } = inObj
if (height === undefined) throw new Error('Block height must be provided')
const options = this.getAxiosOptions()
options.data.id = 'getblockhash'
options.data.method = 'getblockhash'
options.data.params = [parseInt(height)]
const response = await this.axios.request(options)
return response.data.result
}
async getRawTransaction (txid, verbose = true) {
if (!txid) throw new Error('txid must be provided')
const options = this.getAxiosOptions()
options.data.id = 'getrawtransaction'
options.data.method = 'getrawtransaction'
options.data.params = [txid, verbose]
const response = await this.axios.request(options)
return response.data.result
}
}
export default RPC
@@ -0,0 +1,58 @@
/*
Adapter for indexer status in psf-memo-db.
*/
import axios from 'axios'
import RetryQueue from '@chris.troutner/retry-queue'
import config from '../../config/index.js'
import RPC from './rpc.js'
class StatusDb {
constructor () {
this.axios = axios
this.config = config
this.rpc = new RPC()
this.retryQueue = new RetryQueue()
this.getStatus = this.getStatus.bind(this)
this.updateStatus = this.updateStatus.bind(this)
}
async getStatus () {
try {
const response = await this.axios.get(`${this.config.psfMemoDbUrl}/level/status/status`)
return response.data
} catch (err) {
console.log('State not found. Creating fresh state.')
if (this.config.exitOnMissingBackup) {
console.log('EXIT_ON_MISSING_BACKUP set. Exiting.')
process.exit(1)
}
const biggestBlockHeight = await this.retryQueue.addToQueue(this.rpc.getBlockCount, {})
const start = this.config.startBlockHeight - 1
const statusData = {
startBlockHeight: start,
syncedBlockHeight: start,
chainBlockHeight: biggestBlockHeight
}
await this.axios.post(`${this.config.psfMemoDbUrl}/level/status`, {
statusKey: 'status',
statusData
})
return statusData
}
}
async updateStatus (status) {
const { startBlockHeight, syncedBlockHeight, chainBlockHeight } = status
await this.axios.put(`${this.config.psfMemoDbUrl}/level/status`, {
statusData: { startBlockHeight, syncedBlockHeight, chainBlockHeight }
})
return true
}
}
export default StatusDb
@@ -0,0 +1,63 @@
/*
Transaction adapter: fetch TX data and detect Memo OP_RETURN outputs.
*/
import RetryQueue from '@chris.troutner/retry-queue'
import RPC from './rpc.js'
import { findMemoOutputs } from '../lib/memo-parser.js'
import config from '../../config/index.js'
class Transaction {
constructor () {
this.rpc = new RPC()
this.queue = new RetryQueue()
this.config = config
this.txCache = {}
this.txCacheKeys = []
this.get = this.get.bind(this)
this.isMemoTx = this.isMemoTx.bind(this)
this.getTxData = this.getTxData.bind(this)
}
async getTxData (txid) {
const cached = this.txCache[txid]
if (cached) return cached
const txDetails = await this.queue.addToQueue(this.rpc.getRawTransaction, txid)
if (txDetails.blockhash) {
const blockHeader = await this.rpc.getBlockHeader(txDetails.blockhash)
txDetails.blockheight = blockHeader.height
} else {
const blockHeight = await this.rpc.getBlockCount()
txDetails.blockheight = blockHeight + 1
}
this.txCache[txid] = txDetails
this.txCacheKeys.push(txid)
if (this.txCacheKeys.length > this.config.txCacheMax) {
const old = this.txCacheKeys.shift()
delete this.txCache[old]
}
return txDetails
}
async get (txid) {
if (typeof txid !== 'string') {
throw new Error('Input to Transaction.get() must be a string TXID.')
}
return this.getTxData(txid)
}
async isMemoTx (txid) {
try {
const tx = await this.getTxData(txid)
return findMemoOutputs(tx).length > 0
} catch (err) {
return false
}
}
}
export default Transaction
@@ -0,0 +1,23 @@
/*
Trigger TX indexer after block IBD completes.
*/
import axios from 'axios'
import config from '../../config/index.js'
class TxIndexerAdapter {
constructor () {
this.axios = axios
this.config = config
}
async startTxIndexer () {
const response = await this.axios.get(
`http://${this.config.txRestApiIp}:${this.config.txRestApiPort}/tx-start`
)
console.log('TX indexer start response:', response.data)
return true
}
}
export default TxIndexerAdapter
+70
View File
@@ -0,0 +1,70 @@
/*
ZMQ adapter for full node notifications.
*/
import BitcoinCashZmqDecoder from '@psf/bitcoincash-zmq-decoder'
import * as zmq from 'zeromq'
import config from '../../config/index.js'
class ZMQ {
constructor () {
this.sock = new zmq.Subscriber()
this.bchZmqDecoder = new BitcoinCashZmqDecoder('mainnet')
this.config = config
this.txQueue = []
this.blockQueue = []
this.connect = this.connect.bind(this)
this.monitorZmq = this.monitorZmq.bind(this)
this.getTx = this.getTx.bind(this)
this.getBlock = this.getBlock.bind(this)
this.decodeMsg = this.decodeMsg.bind(this)
}
async connect () {
this.sock.connect(`tcp://${this.config.rpcIp}:${this.config.zmqPort}`)
this.sock.subscribe('raw')
this.monitorZmq()
return true
}
async monitorZmq () {
for await (const [topic, msg] of this.sock) {
this.decodeMsg(topic, msg)
}
}
decodeMsg (topic, message) {
try {
const decoded = topic.toString('ascii')
if (decoded === 'rawtx') {
const txd = this.bchZmqDecoder.decodeTransaction(message)
this.txQueue.push(txd.format.txid)
if (this.txQueue.length > this.config.zmqTxQueueMax) {
this.txQueue.shift()
}
} else if (decoded === 'rawblock') {
const blk = this.bchZmqDecoder.decodeBlock(message)
this.blockQueue.push(blk)
if (this.blockQueue.length > this.config.zmqBlockQueueMax) {
this.blockQueue.shift()
}
}
return true
} catch (err) {
console.error('Error in decodeMsg: ', err)
return false
}
}
getTx () {
const nextTx = this.txQueue.shift()
return nextTx === undefined ? false : nextTx
}
getBlock () {
const nextBlock = this.blockQueue.shift()
return nextBlock === undefined ? false : nextBlock
}
}
export default ZMQ
@@ -0,0 +1,25 @@
import Keyboard from './keyboard.js'
import TxRESTController from './tx-rest-api.js'
class Controllers {
constructor (localConfig = {}) {
if (!localConfig.useCases) throw new Error('Use cases required.')
if (!localConfig.adapters) throw new Error('Adapters required.')
this.useCases = localConfig.useCases
this.adapters = localConfig.adapters
this.keyboard = new Keyboard()
this.txRESTController = new TxRESTController(localConfig)
this.initControllers = this.initControllers.bind(this)
this.startTxRESTController = this.startTxRESTController.bind(this)
}
async initControllers () {
this.keyboard.initKeyboard()
}
async startTxRESTController () {
this.txRESTController.start()
}
}
export default Controllers
@@ -0,0 +1,28 @@
import readline from 'readline'
class Keyboard {
constructor () {
this.stopIndexing = false
this.initKeyboard = this.initKeyboard.bind(this)
this.stopStatus = this.stopStatus.bind(this)
}
initKeyboard () {
readline.emitKeypressEvents(process.stdin)
if (process.stdin.isTTY) process.stdin.setRawMode(true)
process.stdin.on('keypress', (str, key) => {
if (key && key.name === 'q') {
this.stopIndexing = true
}
if (key && key.ctrl && key.name === 'c') {
process.exit(0)
}
})
}
stopStatus () {
return this.stopIndexing
}
}
export default Keyboard
@@ -0,0 +1,28 @@
import express from 'express'
import config from '../../config/index.js'
const port = config.txRestApiPort
class TxRESTController {
constructor (localConfig = {}) {
this.adapters = localConfig.adapters
this.useCases = localConfig.useCases
this.app = express()
this.runTxIndexing = false
this.start = this.start.bind(this)
}
start () {
this.app.get('/tx-start', (req, res) => {
this.runTxIndexing = true
console.log('Starting TX Indexer...')
res.send({ report: { success: true } })
})
this.app.listen(port, () => {
console.log(`TX Indexer listening at http://localhost:${port}`)
})
}
}
export default TxRESTController
+33
View File
@@ -0,0 +1,33 @@
/*
Debug logging for Memo indexer (controlled by DEBUG_LEVEL).
*/
import config from '../../config/index.js'
export function getDebugLevel () {
const level = config.debugLevel
if (typeof level !== 'number' || Number.isNaN(level) || level < 0) {
return 0
}
return level
}
/**
* Level 1: per-Memo-tx line with action type(s) and outcome.
*/
export function logMemoTxResult ({ txid, actions = [], success, error, skipped }) {
if (getDebugLevel() < 1) return
const actionStr = actions.length ? actions.join(', ') : 'unknown'
if (skipped) {
console.log(`Memo tx ${txid} actions=[${actionStr}] skipped (already indexed)`)
return
}
if (success) {
console.log(`Memo tx ${txid} actions=[${actionStr}] ok`)
} else {
console.log(`Memo tx ${txid} actions=[${actionStr}] failed: ${error || 'unknown'}`)
}
}
+57
View File
@@ -0,0 +1,57 @@
/*
Memo protocol action codes (mirrors Go ref/bitcoin/memo/codes.go).
*/
export const CODE_PREFIX = 0x6d
export const CODE_SET_NAME = 0x01
export const CODE_POST = 0x02
export const CODE_REPLY = 0x03
export const CODE_LIKE = 0x04
export const CODE_SET_PROFILE = 0x05
export const CODE_FOLLOW = 0x06
export const CODE_UNFOLLOW = 0x07
export const CODE_SET_PROFILE_PIC = 0x0a
export const CODE_TOPIC_MESSAGE = 0x0c
export const CODE_TOPIC_FOLLOW = 0x0d
export const CODE_TOPIC_UNFOLLOW = 0x0e
export const PREFIX_SET_NAME = Buffer.from([CODE_PREFIX, CODE_SET_NAME])
export const PREFIX_POST = Buffer.from([CODE_PREFIX, CODE_POST])
export const PREFIX_REPLY = Buffer.from([CODE_PREFIX, CODE_REPLY])
export const PREFIX_LIKE = Buffer.from([CODE_PREFIX, CODE_LIKE])
export const PREFIX_SET_PROFILE = Buffer.from([CODE_PREFIX, CODE_SET_PROFILE])
export const PREFIX_FOLLOW = Buffer.from([CODE_PREFIX, CODE_FOLLOW])
export const PREFIX_UNFOLLOW = Buffer.from([CODE_PREFIX, CODE_UNFOLLOW])
export const PREFIX_SET_PROFILE_PIC = Buffer.from([CODE_PREFIX, CODE_SET_PROFILE_PIC])
export const PREFIX_TOPIC_MESSAGE = Buffer.from([CODE_PREFIX, CODE_TOPIC_MESSAGE])
export const PREFIX_TOPIC_FOLLOW = Buffer.from([CODE_PREFIX, CODE_TOPIC_FOLLOW])
export const PREFIX_TOPIC_UNFOLLOW = Buffer.from([CODE_PREFIX, CODE_TOPIC_UNFOLLOW])
export const MAX_POST_SIZE = 65000
export const MAX_REPLY_SIZE = 65000
export const TX_HASH_LENGTH = 32
export const PK_HASH_LENGTH = 20
export const ACTION_NAMES = {
[`${CODE_PREFIX}-${CODE_SET_NAME}`]: 'setName',
[`${CODE_PREFIX}-${CODE_POST}`]: 'post',
[`${CODE_PREFIX}-${CODE_REPLY}`]: 'reply',
[`${CODE_PREFIX}-${CODE_LIKE}`]: 'like',
[`${CODE_PREFIX}-${CODE_SET_PROFILE}`]: 'setProfile',
[`${CODE_PREFIX}-${CODE_FOLLOW}`]: 'follow',
[`${CODE_PREFIX}-${CODE_UNFOLLOW}`]: 'unfollow',
[`${CODE_PREFIX}-${CODE_SET_PROFILE_PIC}`]: 'setProfilePic',
[`${CODE_PREFIX}-${CODE_TOPIC_MESSAGE}`]: 'topicMessage',
[`${CODE_PREFIX}-${CODE_TOPIC_FOLLOW}`]: 'topicFollow',
[`${CODE_PREFIX}-${CODE_TOPIC_UNFOLLOW}`]: 'topicUnfollow'
}
export function isMemoPrefix (buf) {
return buf && buf.length >= 2 && buf[0] === CODE_PREFIX
}
export function getActionFromPrefix (prefixBuf) {
if (!isMemoPrefix(prefixBuf)) return null
return ACTION_NAMES[`${prefixBuf[0]}-${prefixBuf[1]}`] || null
}
+185
View File
@@ -0,0 +1,185 @@
/*
Parse Memo OP_RETURN scripts and extract signer addresses from transactions.
*/
import BCHJS from '@psf/bch-js'
import {
isMemoPrefix,
getActionFromPrefix,
CODE_PREFIX
} from './memo-codes.js'
let bchjs
function getBchjs () {
if (!bchjs) {
bchjs = new BCHJS({
restURL: process.env.RESTURL || 'https://api.fullstack.cash/v5/'
})
}
return bchjs
}
/**
* Parse push data chunks from a script hex string.
*/
export function parseScriptPushDatas (scriptHex) {
if (!scriptHex || typeof scriptHex !== 'string') return []
const buf = Buffer.from(scriptHex, 'hex')
const pushDatas = []
let i = 0
while (i < buf.length) {
const op = buf[i]
if (op === 0x00) {
pushDatas.push(Buffer.alloc(0))
i++
continue
}
if (op >= 0x01 && op <= 0x4b) {
const len = op
pushDatas.push(buf.slice(i + 1, i + 1 + len))
i += 1 + len
continue
}
if (op === 0x4c) {
const len = buf[i + 1]
pushDatas.push(buf.slice(i + 2, i + 2 + len))
i += 2 + len
continue
}
if (op === 0x4d) {
const len = buf.readUInt16LE(i + 1)
pushDatas.push(buf.slice(i + 3, i + 3 + len))
i += 3 + len
continue
}
if (op === 0x4e) {
const len = buf.readUInt32LE(i + 1)
pushDatas.push(buf.slice(i + 5, i + 5 + len))
i += 5 + len
continue
}
// Skip opcode (OP_RETURN 0x6a, etc.)
i++
}
return pushDatas
}
/**
* Decode Memo OP_RETURN from a scriptPubKey hex string.
*/
export function decodeMemoOpReturn (scriptHex) {
const pushDatas = parseScriptPushDatas(scriptHex)
if (!pushDatas.length) return null
// OP_RETURN scripts: first push may be after OP_RETURN opcode parsing;
// find first memo prefix in pushes
for (const data of pushDatas) {
if (!isMemoPrefix(data)) continue
const action = getActionFromPrefix(data)
if (!action) continue
return { action, prefix: data, pushDatas }
}
return null
}
export function isMemoScript (scriptHex) {
return decodeMemoOpReturn(scriptHex) !== null
}
export function findMemoOutputs (txDetails) {
const matches = []
if (!txDetails || !txDetails.vout) return matches
for (let i = 0; i < txDetails.vout.length; i++) {
const vout = txDetails.vout[i]
const hex = vout.scriptPubKey && vout.scriptPubKey.hex
const decoded = decodeMemoOpReturn(hex)
if (decoded) {
matches.push({ voutIndex: i, ...decoded })
}
}
return matches
}
/**
* Extract compressed or uncompressed pubkey from a P2PKH unlocking script.
* Mirrors Go wallet.GetP2pkhAddrFromUnlockScript (pubkey push after signature).
*/
export function getPubkeyFromUnlockScript (scriptBuf) {
if (!scriptBuf || !scriptBuf.length) return null
let decoded
try {
decoded = getBchjs().Script.decode(scriptBuf)
} catch (err) {
return null
}
if (!decoded || !decoded.length) return null
for (let i = decoded.length - 1; i >= 0; i--) {
const chunk = decoded[i]
if (Buffer.isBuffer(chunk) && (chunk.length === 33 || chunk.length === 65)) {
return chunk
}
}
return null
}
/**
* Derive cash address from unlocking script hex (P2PKH spend).
*/
export function getAddressFromUnlockScriptHex (scriptHex) {
if (!scriptHex || typeof scriptHex !== 'string') return null
try {
const scriptBuf = Buffer.from(scriptHex, 'hex')
const pubkey = getPubkeyFromUnlockScript(scriptBuf)
if (!pubkey) return null
const hash160 = getBchjs().Crypto.hash160(pubkey)
return getBchjs().Address.hash160ToCash(hash160)
} catch (err) {
return null
}
}
export function getSignerAddress (txDetails) {
if (!txDetails || !txDetails.vin) return null
for (const vin of txDetails.vin) {
const scriptHex = vin.scriptSig && vin.scriptSig.hex
if (!scriptHex) continue
const addr = getAddressFromUnlockScriptHex(scriptHex)
if (addr) return addr
}
return null
}
export function getPkHashFromAddress (address) {
try {
const decoded = getBchjs().Address.decode(address)
return Buffer.from(decoded.hash)
} catch (err) {
return null
}
}
export function prefixToHex (prefixBuf) {
return prefixBuf.toString('hex')
}
export { CODE_PREFIX, isMemoPrefix }
@@ -0,0 +1,30 @@
import { logProcessError } from './helpers.js'
import { PK_HASH_LENGTH, PREFIX_UNFOLLOW } from '../../lib/memo-codes.js'
export async function handleFollow (ctx) {
const { adapters, txid, signerAddr, decoded, seen, blockHeight } = ctx
const { pushDatas, prefix } = decoded
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid follow push data count ${pushDatas.length}`, blockHeight)
return
}
if (pushDatas[1].length !== PK_HASH_LENGTH) {
await logProcessError(adapters, txid, 'follow pk hash wrong size', blockHeight)
return
}
const unfollow = prefix[1] === PREFIX_UNFOLLOW[1]
const followeePkHash = pushDatas[1].toString('hex')
const key = `${signerAddr}:${followeePkHash}`
await adapters.followDb.create(key, {
followerAddr: signerAddr,
followeePkHash,
unfollow,
txid,
seen,
blockHeight
})
}
@@ -0,0 +1,67 @@
/*
Shared helpers for Memo action handlers.
*/
import { isMemoPrefix } from '../../lib/memo-codes.js'
export async function logProcessError (adapters, txid, error, blockHeight) {
try {
await adapters.processErrorDb.create(txid, { error, ts: Date.now(), blockHeight })
} catch (err) {
console.error('Failed to log process error:', err.message)
}
}
export function utf8FromPush (buf) {
return buf.toString('utf8')
}
/**
* Drop leading empty pushes (btcd txscript.PushedData compatibility).
*/
export function stripLeadingEmptyPushes (pushDatas) {
let datas = pushDatas
while (datas.length > 1 && datas[0] && datas[0].length === 0) {
datas = datas.slice(1)
}
return datas
}
/**
* Normalize Memo actions that use prefix + payload as two script pushes.
* Some wallets encode both in a single push (0x6dXX + data); split for handlers.
*/
export function normalizeTwoPushMemoDatas (pushDatas) {
const datas = stripLeadingEmptyPushes(pushDatas)
if (datas.length === 2) {
return datas
}
if (datas.length === 1 && isMemoPrefix(datas[0]) && datas[0].length > 2) {
return [datas[0].subarray(0, 2), datas[0].subarray(2)]
}
return datas
}
export function txHashFromPush (buf) {
if (!buf || buf.length !== 32) return null
return Buffer
.from(buf)
.reverse()
.toString('hex')
}
export function followKey (followerAddr, followeeAddr) {
return `${followerAddr}:${followeeAddr}`
}
export function roomKey (roomName, txid) {
return `${roomName}:${txid}`
}
export function postChildKey (parentTxid, childTxid) {
return `${parentTxid}:${childTxid}`
}
@@ -0,0 +1,33 @@
import { handleSetName } from './set-name.js'
import { handlePost } from './post.js'
import { handleReply } from './reply.js'
import { handleLike } from './like.js'
import { handleSetProfile } from './set-profile.js'
import { handleFollow } from './follow.js'
import { handleSetProfilePic } from './set-profile-pic.js'
import { handleTopicMessage } from './topic-message.js'
import { handleTopicFollow } from './topic-follow.js'
export const ACTION_HANDLERS = {
setName: handleSetName,
post: handlePost,
reply: handleReply,
like: handleLike,
setProfile: handleSetProfile,
follow: handleFollow,
unfollow: handleFollow,
setProfilePic: handleSetProfilePic,
topicMessage: handleTopicMessage,
topicFollow: handleTopicFollow,
topicUnfollow: handleTopicFollow
}
export async function dispatchMemoAction (ctx) {
const handler = ACTION_HANDLERS[ctx.decoded.action]
if (!handler) {
console.log(`No handler for action ${ctx.decoded.action}`)
return false
}
await handler(ctx)
return true
}
@@ -0,0 +1,43 @@
import { txHashFromPush, logProcessError } from './helpers.js'
import { TX_HASH_LENGTH } from '../../lib/memo-codes.js'
export async function handleLike (ctx) {
const { adapters, txid, signerAddr, decoded, seen, txDetails, blockHeight } = ctx
const { pushDatas } = decoded
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid like push data count ${pushDatas.length}`, blockHeight)
return
}
if (pushDatas[1].length !== TX_HASH_LENGTH) {
await logProcessError(adapters, txid, 'like post tx hash wrong size', blockHeight)
return
}
const postTxid = txHashFromPush(pushDatas[1])
let tip = 0
try {
const post = await adapters.postDb.get(postTxid)
if (post && post.addr !== signerAddr) {
for (const vout of txDetails.vout) {
const addrs = vout.scriptPubKey && vout.scriptPubKey.addresses
if (addrs && addrs[0] === post.addr) {
tip += Math.round(vout.value * 1e8)
}
}
}
} catch (err) {
// post may not exist yet
}
const likeData = {
addr: signerAddr,
postTxid,
seen,
tip,
blockHeight
}
await adapters.likeDb.create(txid, likeData)
}
@@ -0,0 +1,29 @@
import { utf8FromPush, logProcessError, normalizeTwoPushMemoDatas } from './helpers.js'
import { MAX_POST_SIZE } from '../../lib/memo-codes.js'
export async function handlePost (ctx) {
const { adapters, txid, signerAddr, decoded, seen, blockHeight } = ctx
const pushDatas = normalizeTwoPushMemoDatas(decoded.pushDatas)
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid post push data count ${pushDatas.length}`, blockHeight)
return
}
const text = utf8FromPush(pushDatas[1])
if (!text.length) {
await logProcessError(adapters, txid, 'empty post', blockHeight)
return
}
if (text.length > MAX_POST_SIZE) {
await logProcessError(adapters, txid, 'post too large', blockHeight)
return
}
const postData = { addr: signerAddr, text, seen, blockHeight }
try {
await adapters.postDb.get(txid)
} catch (err) {
await adapters.postDb.create(txid, postData)
}
}
@@ -0,0 +1,30 @@
import { utf8FromPush, txHashFromPush, logProcessError, postChildKey } from './helpers.js'
import { MAX_REPLY_SIZE } from '../../lib/memo-codes.js'
import { handlePost } from './post.js'
export async function handleReply (ctx) {
const { adapters, txid, decoded, blockHeight } = ctx
const { pushDatas } = decoded
if (pushDatas.length !== 3) {
await logProcessError(adapters, txid, `invalid reply push data count ${pushDatas.length}`, blockHeight)
return
}
const parentTxid = txHashFromPush(pushDatas[1])
if (!parentTxid) {
await logProcessError(adapters, txid, 'invalid parent tx hash for reply', blockHeight)
return
}
const text = utf8FromPush(pushDatas[2])
if (text.length > MAX_REPLY_SIZE) {
await logProcessError(adapters, txid, 'reply too large', blockHeight)
return
}
await adapters.postParentDb.create(txid, { parentTxid, childTxid: txid, blockHeight })
await adapters.postChildDb.create(postChildKey(parentTxid, txid), { parentTxid, childTxid: txid, blockHeight })
await handlePost({ ...ctx, decoded: { ...decoded, pushDatas: [pushDatas[0], pushDatas[2]] } })
}
@@ -0,0 +1,20 @@
import { utf8FromPush, logProcessError, normalizeTwoPushMemoDatas } from './helpers.js'
import { MAX_POST_SIZE } from '../../lib/memo-codes.js'
export async function handleSetName (ctx) {
const { adapters, txid, signerAddr, decoded, seen, blockHeight } = ctx
const pushDatas = normalizeTwoPushMemoDatas(decoded.pushDatas)
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid set name push data count ${pushDatas.length}`, blockHeight)
return
}
const name = utf8FromPush(pushDatas[1])
if (name.length > MAX_POST_SIZE) {
await logProcessError(adapters, txid, 'set name too large', blockHeight)
return
}
await adapters.nameDb.create(signerAddr, { name, txid, seen, addr: signerAddr, blockHeight })
}
@@ -0,0 +1,24 @@
import { utf8FromPush, logProcessError, normalizeTwoPushMemoDatas } from './helpers.js'
import { MAX_POST_SIZE } from '../../lib/memo-codes.js'
export async function handleSetProfilePic (ctx) {
const { adapters, txid, signerAddr, decoded, seen, blockHeight } = ctx
const pushDatas = normalizeTwoPushMemoDatas(decoded.pushDatas)
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid profile pic push data count ${pushDatas.length}`, blockHeight)
return
}
const url = utf8FromPush(pushDatas[1])
if (!url.length) {
await logProcessError(adapters, txid, 'empty profile pic url', blockHeight)
return
}
if (url.length > MAX_POST_SIZE) {
await logProcessError(adapters, txid, 'profile pic url too large', blockHeight)
return
}
await adapters.profilePicDb.create(signerAddr, { url, txid, seen, addr: signerAddr, blockHeight })
}
@@ -0,0 +1,20 @@
import { utf8FromPush, logProcessError, normalizeTwoPushMemoDatas } from './helpers.js'
import { MAX_POST_SIZE } from '../../lib/memo-codes.js'
export async function handleSetProfile (ctx) {
const { adapters, txid, signerAddr, decoded, seen, blockHeight } = ctx
const pushDatas = normalizeTwoPushMemoDatas(decoded.pushDatas)
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid profile push data count ${pushDatas.length}`, blockHeight)
return
}
const text = utf8FromPush(pushDatas[1])
if (text.length > MAX_POST_SIZE) {
await logProcessError(adapters, txid, 'profile too large', blockHeight)
return
}
await adapters.profileDb.create(signerAddr, { text, txid, seen, addr: signerAddr, blockHeight })
}
@@ -0,0 +1,25 @@
import { utf8FromPush, logProcessError, roomKey } from './helpers.js'
import { PREFIX_TOPIC_UNFOLLOW } from '../../lib/memo-codes.js'
export async function handleTopicFollow (ctx) {
const { adapters, txid, signerAddr, decoded, seen, blockHeight } = ctx
const { pushDatas, prefix } = decoded
if (pushDatas.length !== 2) {
await logProcessError(adapters, txid, `invalid topic follow push data count ${pushDatas.length}`, blockHeight)
return
}
const room = utf8FromPush(pushDatas[1])
const unfollow = prefix[1] === PREFIX_TOPIC_UNFOLLOW[1]
await adapters.roomDb.create(roomKey(room, signerAddr), {
room,
addr: signerAddr,
unfollow,
txid,
seen,
type: 'follow',
blockHeight
})
}
@@ -0,0 +1,27 @@
import { utf8FromPush, logProcessError, roomKey } from './helpers.js'
import { MAX_POST_SIZE } from '../../lib/memo-codes.js'
import { handlePost } from './post.js'
export async function handleTopicMessage (ctx) {
const { adapters, txid, decoded, seen, blockHeight } = ctx
const { pushDatas } = decoded
if (pushDatas.length !== 3) {
await logProcessError(adapters, txid, `invalid topic message push data count ${pushDatas.length}`, blockHeight)
return
}
const room = utf8FromPush(pushDatas[1])
const message = utf8FromPush(pushDatas[2])
if ((room.length + message.length) > MAX_POST_SIZE) {
await logProcessError(adapters, txid, 'topic message too large', blockHeight)
return
}
await handlePost({
...ctx,
decoded: { ...decoded, pushDatas: [pushDatas[0], pushDatas[2]] }
})
await adapters.roomDb.create(roomKey(room, txid), { room, txid, seen, type: 'post', blockHeight })
}
@@ -0,0 +1,45 @@
/*
Filter block transactions for Memo OP_RETURN outputs.
*/
import PQueue from 'p-queue'
import pRetry from 'p-retry'
import config from '../../config/index.js'
class FilterBlock {
constructor (localConfig = {}) {
if (!localConfig.adapters) {
throw new Error('Adapters required for filter-block.js')
}
this.adapters = localConfig.adapters
this.pQueue = new PQueue({ concurrency: config.filterConcurrency })
this.pRetry = pRetry
this.attempts = 5
this.filterMemoTxs = this.filterMemoTxs.bind(this)
}
async retryWrapper (funcHandle, inputObj) {
return this.pRetry(async () => funcHandle(inputObj), {
retries: this.attempts,
onFailedAttempt: (error) => {
console.log(`Attempt ${error.attemptNumber} failed. ${error.retriesLeft} retries left.`)
}
})
}
async filterMemoTxs (txids) {
const memoTxSet = new Set()
const tasks = txids.map((txid) => async () => {
const isMemo = await this.retryWrapper(
this.adapters.transaction.isMemoTx.bind(this.adapters.transaction),
txid
)
if (isMemo) memoTxSet.add(txid)
})
await this.pQueue.addAll(tasks)
return txids.filter((txid) => memoTxSet.has(txid))
}
}
export default FilterBlock
@@ -0,0 +1,129 @@
/*
Business logic for indexing blocks with Memo transactions.
*/
import RetryQueue from '@chris.troutner/retry-queue'
import PQueue from 'p-queue'
import config from '../../config/index.js'
import FilterBlock from './filter-block.js'
import { findMemoOutputs, getSignerAddress } from '../lib/memo-parser.js'
import { logMemoTxResult } from '../lib/debug-log.js'
import { dispatchMemoAction } from './action-types/index.js'
class IndexBlocks {
constructor (localConfig = {}) {
if (!localConfig.adapters) {
throw new Error('Adapters required for index-blocks.js')
}
this.adapters = localConfig.adapters
this.filterBlock = new FilterBlock({ adapters: this.adapters })
this.retryQueue = new RetryQueue()
this.pQueue = new PQueue({ concurrency: config.memoTxConcurrency })
this.processBlock = this.processBlock.bind(this)
this.processMemoTx = this.processMemoTx.bind(this)
this.processMemoTxs = this.processMemoTxs.bind(this)
}
async getProcessErrorMessage (txid) {
try {
const record = await this.adapters.processErrorDb.get(txid)
return record && record.error ? record.error : null
} catch (err) {
return null
}
}
async processMemoTx (txid, blockHeight, seen = Date.now()) {
let actions = []
try {
try {
await this.adapters.ptxDb.get(txid)
logMemoTxResult({ txid, actions, skipped: true, success: true })
return true
} catch (err) {
// not processed yet
}
const txDetails = await this.adapters.transaction.get(txid)
const memoOutputs = findMemoOutputs(txDetails)
actions = memoOutputs.map((output) => output.action)
const signerAddr = getSignerAddress(txDetails)
if (!signerAddr) {
const error = 'could not find input address for memo tx'
await this.adapters.processErrorDb.create(txid, {
error,
ts: Date.now()
})
logMemoTxResult({ txid, actions, success: false, error })
return false
}
const processedAt = Date.now()
for (const decoded of memoOutputs) {
await dispatchMemoAction({
adapters: this.adapters,
txid,
txDetails,
signerAddr,
decoded,
seen,
blockHeight
})
}
const handlerError = await this.getProcessErrorMessage(txid)
if (handlerError) {
logMemoTxResult({ txid, actions, success: false, error: handlerError })
await this.adapters.ptxDb.create(txid, { processedAt, blockHeight })
return true
}
await this.adapters.ptxDb.create(txid, { processedAt, blockHeight })
logMemoTxResult({ txid, actions, success: true })
return true
} catch (err) {
logMemoTxResult({ txid, actions, success: false, error: err.message })
console.error(`Error processing memo tx ${txid}:`, err.message)
throw err
}
}
async processMemoTxs (txids, blockHeight, seen) {
const tasks = txids.map((txid) => async () => {
await this.processMemoTx(txid, blockHeight, seen)
})
await this.pQueue.addAll(tasks)
return true
}
async processBlock (blockHeight) {
const blockHash = await this.retryQueue.addToQueue(
this.adapters.rpc.getBlockHash,
{ height: blockHeight }
)
const block = await this.retryQueue.addToQueue(
this.adapters.rpc.getBlock,
{ hash: blockHash }
)
const txs = block.tx
const now = new Date()
console.log(
`\nIndexing block ${blockHeight} with ${txs.length} txs. ${now.toLocaleString()}`
)
const memoTxs = await this.filterBlock.filterMemoTxs(txs)
if (memoTxs.length) {
console.log(`Memo txs in block: ${memoTxs.length}`)
const blockSeen = block.time * 1000
await this.processMemoTxs(memoTxs, blockHeight, blockSeen)
}
return true
}
}
export default IndexBlocks
+34
View File
@@ -0,0 +1,34 @@
import RetryQueue from '@chris.troutner/retry-queue'
class State {
constructor (localConfig = {}) {
if (!localConfig.adapters) {
throw new Error('Adapters required for state.js')
}
this.adapters = localConfig.adapters
this.retryQueue = new RetryQueue()
this.getStatus = this.getStatus.bind(this)
this.updateIndexedBlockHeight = this.updateIndexedBlockHeight.bind(this)
}
async getStatus () {
return this.adapters.statusDb.getStatus()
}
async updateIndexedBlockHeight (inObj = {}) {
const { lastIndexedBlockHeight } = inObj
const status = await this.adapters.statusDb.getStatus()
if (status.syncedBlockHeight !== (lastIndexedBlockHeight - 1)) {
throw new Error(
`Expected synced block height ${lastIndexedBlockHeight - 1}, got ${status.syncedBlockHeight}`
)
}
status.syncedBlockHeight = lastIndexedBlockHeight
await this.adapters.statusDb.updateStatus(status)
return lastIndexedBlockHeight + 1
}
}
export default State
@@ -0,0 +1,25 @@
import RetryQueue from '@chris.troutner/retry-queue'
import IndexBlocks from './index-blocks.js'
import State from './state.js'
import Utils from './utils.js'
class UseCases {
constructor (localConfig = {}) {
if (!localConfig.adapters) {
throw new Error('Adapters required for use cases.')
}
this.adapters = localConfig.adapters
this.indexBlocks = new IndexBlocks({ adapters: this.adapters })
this.state = new State({ adapters: this.adapters })
this.utils = new Utils()
this.retryQueue = new RetryQueue()
this.initUseCases = this.initUseCases.bind(this)
}
async initUseCases () {
console.log('Use cases initialized.')
return true
}
}
export default UseCases
+7
View File
@@ -0,0 +1,7 @@
class Utils {
sleep (ms = 1000) {
return new Promise((resolve) => setTimeout(resolve, ms))
}
}
export default Utils
@@ -0,0 +1,29 @@
import { assert } from 'chai'
import sinon from 'sinon'
import Transaction from '../../../src/adapters/transaction.js'
describe('#Transaction', () => {
let uut
let sandbox
beforeEach(() => {
sandbox = sinon.createSandbox()
uut = new Transaction()
})
afterEach(() => sandbox.restore())
it('should detect memo tx', async () => {
sandbox.stub(uut, 'getTxData').resolves({
vout: [{
scriptPubKey: {
hex: '6a026d020474657374'
}
}],
vin: []
})
const result = await uut.isMemoTx('testtx')
assert.equal(result, true)
})
})
@@ -0,0 +1,62 @@
import { assert } from 'chai'
import sinon from 'sinon'
import config from '../../../config/index.js'
import { getDebugLevel, logMemoTxResult } from '../../../src/lib/debug-log.js'
describe('#debug-log', () => {
let sandbox
let originalDebugLevel
beforeEach(() => {
sandbox = sinon.createSandbox()
originalDebugLevel = config.debugLevel
sandbox.stub(console, 'log')
})
afterEach(() => {
config.debugLevel = originalDebugLevel
sandbox.restore()
})
describe('#getDebugLevel', () => {
it('should default invalid levels to 0', () => {
config.debugLevel = NaN
assert.equal(getDebugLevel(), 0)
})
})
describe('#logMemoTxResult', () => {
it('should not log when debug level is 0', () => {
config.debugLevel = 0
logMemoTxResult({
txid: 'abc',
actions: ['post'],
success: true
})
assert.equal(console.log.callCount, 0)
})
it('should log ok with actions at level 1', () => {
config.debugLevel = 1
logMemoTxResult({
txid: 'abc',
actions: ['post'],
success: true
})
assert.include(console.log.firstCall.args[0], 'actions=[post]')
assert.include(console.log.firstCall.args[0], 'ok')
})
it('should log failure with process error at level 1', () => {
config.debugLevel = 1
logMemoTxResult({
txid: 'abc',
actions: ['like'],
success: false,
error: 'invalid like push data count 1'
})
assert.include(console.log.firstCall.args[0], 'failed:')
assert.include(console.log.firstCall.args[0], 'invalid like')
})
})
})
@@ -0,0 +1,88 @@
import { assert } from 'chai'
import {
decodeMemoOpReturn,
parseScriptPushDatas,
isMemoScript,
getSignerAddress,
getAddressFromUnlockScriptHex
} from '../../../src/lib/memo-parser.js'
import { CODE_POST, CODE_PREFIX } from '../../../src/lib/memo-codes.js'
// OP_RETURN post from Go post_test.go: 6d02 + "Post message"
const POST_OP_RETURN_HEX = '6a026d020c506f7374206d657373616765'
describe('#memo-parser', () => {
describe('#parseScriptPushDatas', () => {
it('should parse OP_RETURN push data', () => {
const pushDatas = parseScriptPushDatas(POST_OP_RETURN_HEX)
assert.equal(pushDatas.length, 2)
assert.equal(pushDatas[0][0], CODE_PREFIX)
assert.equal(pushDatas[0][1], CODE_POST)
assert.equal(pushDatas[1].toString('utf8'), 'Post message')
})
})
describe('#decodeMemoOpReturn', () => {
it('should decode a memo post script', () => {
const decoded = decodeMemoOpReturn(POST_OP_RETURN_HEX)
assert.equal(decoded.action, 'post')
assert.equal(decoded.pushDatas[1].toString('utf8'), 'Post message')
})
it('should return null for non-memo script', () => {
const decoded = decodeMemoOpReturn('76a91400')
assert.equal(decoded, null)
})
})
describe('#isMemoScript', () => {
it('should detect memo script', () => {
assert.equal(isMemoScript(POST_OP_RETURN_HEX), true)
})
})
describe('#getAddressFromUnlockScriptHex', () => {
const SCRIPT_SIG_1 =
'483045022100e15e18048b78e744fe8e440eb2687b5d4eb7009140d5e71950d4351c6e3bf2ef022031b0dab7326846a58e16f6a753e0d011514f827d2adacde63912daa3f85636c5412103833927b98fab40d5f857214d802ba42ea21087105b81e9bff57bf2486c9ff171'
const SCRIPT_SIG_2 =
'483045022100aa35ee315859b63f92e7878304364635943b28b2334dfd796a51cb2e0b284fe3022033e852b7c965f36c7d6eecc577b01218d611450c80e75af1fd41e76cfc7af81c4121035c5970c98aa4543271348bb18de4fb04122af87555fc4411c376493cc85594c3'
it('should derive cash address from real memo tx scriptSig', () => {
const addr = getAddressFromUnlockScriptHex(SCRIPT_SIG_1)
assert.isString(addr)
assert.include(addr, 'bitcoincash:')
assert.equal(
addr,
'bitcoincash:qzd7s66p0mh97s2tqkrwacx4dazwataul5m9f9yw68'
)
})
it('should derive cash address from second sample scriptSig', () => {
const addr = getAddressFromUnlockScriptHex(SCRIPT_SIG_2)
assert.equal(
addr,
'bitcoincash:qrnr8q2ndfmzgfehz5txel9jpq3lrwhvt5f270fnd4'
)
})
})
describe('#getSignerAddress', () => {
const SCRIPT_SIG_1 =
'483045022100e15e18048b78e744fe8e440eb2687b5d4eb7009140d5e71950d4351c6e3bf2ef022031b0dab7326846a58e16f6a753e0d011514f827d2adacde63912daa3f85636c5412103833927b98fab40d5f857214d802ba42ea21087105b81e9bff57bf2486c9ff171'
it('should return signer from txDetails with vin scriptSig', () => {
const txDetails = {
vin: [{ scriptSig: { hex: SCRIPT_SIG_1 } }]
}
const addr = getSignerAddress(txDetails)
assert.equal(
addr,
'bitcoincash:qzd7s66p0mh97s2tqkrwacx4dazwataul5m9f9yw68'
)
})
it('should return null when no vin', () => {
assert.equal(getSignerAddress({}), null)
})
})
})
@@ -0,0 +1,71 @@
import { assert } from 'chai'
import sinon from 'sinon'
import {
normalizeTwoPushMemoDatas,
stripLeadingEmptyPushes,
logProcessError
} from '../../../../src/use-cases/action-types/helpers.js'
import { PREFIX_SET_PROFILE_PIC, PREFIX_POST } from '../../../../src/lib/memo-codes.js'
describe('#action-types/helpers', () => {
describe('#normalizeTwoPushMemoDatas', () => {
it('should pass through standard two-push encoding', () => {
const prefix = PREFIX_SET_PROFILE_PIC
const url = Buffer.from('https://example.com/pic.png', 'utf8')
const normalized = normalizeTwoPushMemoDatas([prefix, url])
assert.equal(normalized.length, 2)
assert.equal(normalized[1].toString('utf8'), 'https://example.com/pic.png')
})
it('should split single-push prefix+payload (profile pic on-chain format)', () => {
const url = Buffer.from('-hash-or-url-bytes', 'utf8')
const combined = Buffer.concat([PREFIX_SET_PROFILE_PIC, url])
const normalized = normalizeTwoPushMemoDatas([combined])
assert.equal(normalized.length, 2)
assert.deepEqual(normalized[0], PREFIX_SET_PROFILE_PIC)
assert.deepEqual(normalized[1], url)
})
it('should handle user sample: one push with 0x6d0a prefix and 18-byte payload', () => {
const combined = Buffer.from(
'6d0a2da40232abb64740e8abeef1f102bdc92ed5',
'hex'
)
const normalized = normalizeTwoPushMemoDatas([combined])
assert.equal(normalized.length, 2)
assert.equal(normalized[0][1], 0x0a)
assert.equal(normalized[1].length, 18)
})
it('should strip leading empty push before normalizing', () => {
const message = Buffer.from('hello', 'utf8')
const combined = Buffer.concat([PREFIX_POST, message])
const normalized = normalizeTwoPushMemoDatas([Buffer.alloc(0), combined])
assert.equal(normalized.length, 2)
assert.equal(normalized[1].toString('utf8'), 'hello')
})
})
describe('#stripLeadingEmptyPushes', () => {
it('should remove only leading empty pushes', () => {
const payload = Buffer.from([0xab])
const result = stripLeadingEmptyPushes([Buffer.alloc(0), payload])
assert.equal(result.length, 1)
assert.deepEqual(result[0], payload)
})
})
describe('#logProcessError', () => {
it('should store blockHeight on process error records', async () => {
const create = sinon.stub().resolves()
const adapters = { processErrorDb: { create } }
await logProcessError(adapters, 'tx1', 'bad data', 600100)
assert.equal(create.callCount, 1)
assert.equal(create.firstCall.args[0], 'tx1')
assert.equal(create.firstCall.args[1].error, 'bad data')
assert.equal(create.firstCall.args[1].blockHeight, 600100)
})
})
})
@@ -0,0 +1,35 @@
import { assert } from 'chai'
import sinon from 'sinon'
import { handlePost } from '../../../../src/use-cases/action-types/post.js'
import { PREFIX_POST } from '../../../../src/lib/memo-codes.js'
describe('#handlePost', () => {
it('should save a post to the database', async () => {
const create = sinon.stub().resolves({ success: true })
const get = sinon.stub().rejects(new Error('not found'))
const adapters = {
postDb: { create, get },
processErrorDb: { create: sinon.stub() }
}
const message = Buffer.from('hello memo')
await handlePost({
adapters,
txid: 'abc123',
signerAddr: 'bitcoincash:qptest',
seen: 1000,
blockHeight: 600100,
decoded: {
action: 'post',
prefix: PREFIX_POST,
pushDatas: [PREFIX_POST, message]
}
})
assert.equal(create.callCount, 1)
assert.equal(create.firstCall.args[0], 'abc123')
assert.equal(create.firstCall.args[1].text, 'hello memo')
assert.equal(create.firstCall.args[1].blockHeight, 600100)
})
})
@@ -0,0 +1,36 @@
import { assert } from 'chai'
import sinon from 'sinon'
import { handleSetProfilePic } from '../../../../src/use-cases/action-types/set-profile-pic.js'
import { PREFIX_SET_PROFILE_PIC } from '../../../../src/lib/memo-codes.js'
describe('#handleSetProfilePic', () => {
let adapters
beforeEach(() => {
adapters = {
profilePicDb: { create: sinon.stub().resolves() },
processErrorDb: { create: sinon.stub().resolves() }
}
})
it('should index profile pic from single-push OP_RETURN', async () => {
const urlBytes = Buffer.from('https://memo.cash/img/abc', 'utf8')
const combined = Buffer.concat([PREFIX_SET_PROFILE_PIC, urlBytes])
await handleSetProfilePic({
adapters,
txid: 'abc123',
signerAddr: 'bitcoincash:qptest',
seen: 12345,
blockHeight: 600200,
decoded: { pushDatas: [combined] }
})
assert.equal(adapters.processErrorDb.create.callCount, 0)
assert.equal(adapters.profilePicDb.create.callCount, 1)
const [addr, data] = adapters.profilePicDb.create.firstCall.args
assert.equal(addr, 'bitcoincash:qptest')
assert.equal(data.url, 'https://memo.cash/img/abc')
assert.equal(data.blockHeight, 600200)
})
})
@@ -0,0 +1,41 @@
import { assert } from 'chai'
import sinon from 'sinon'
import FilterBlock from '../../../src/use-cases/filter-block.js'
describe('#FilterBlock', () => {
let uut
let sandbox
let adapters
beforeEach(() => {
sandbox = sinon.createSandbox()
adapters = {
transaction: {
isMemoTx: sandbox.stub()
}
}
adapters.transaction.isMemoTx.onCall(0).resolves(true)
adapters.transaction.isMemoTx.onCall(1).resolves(false)
adapters.transaction.isMemoTx.onCall(2).resolves(true)
uut = new FilterBlock({ adapters })
})
afterEach(() => sandbox.restore())
it('should filter memo txs from block tx list', async () => {
const result = await uut.filterMemoTxs(['tx1', 'tx2', 'tx3'])
assert.deepEqual(result, ['tx1', 'tx3'])
})
it('should preserve block tx order when filtering in parallel', async () => {
adapters.transaction.isMemoTx.reset()
adapters.transaction.isMemoTx.callsFake(async (txid) => {
return txid === 'tx-a' || txid === 'tx-c' || txid === 'tx-e'
})
const blockOrder = ['tx-a', 'tx-b', 'tx-c', 'tx-d', 'tx-e']
const result = await uut.filterMemoTxs(blockOrder)
assert.deepEqual(result, ['tx-a', 'tx-c', 'tx-e'])
})
})
@@ -0,0 +1,95 @@
import { assert } from 'chai'
import sinon from 'sinon'
import IndexBlocks from '../../../src/use-cases/index-blocks.js'
describe('#IndexBlocks', () => {
let uut
let sandbox
let adapters
beforeEach(() => {
sandbox = sinon.createSandbox()
adapters = {
ptxDb: {
get: sandbox.stub().rejects(new Error('not found')),
create: sandbox.stub().resolves({ success: true })
},
processErrorDb: { create: sandbox.stub().resolves({ success: true }) },
transaction: {
get: sandbox.stub().resolves({
txid: 'tx1',
vin: [{ txid: 'prev', vout: 0, scriptSig: { hex: '471044...' } }],
vout: [{
scriptPubKey: {
hex: '6a046d020102',
addresses: ['bitcoincash:qptest']
}
}]
})
},
rpc: {
getBlockHash: sandbox.stub().resolves('block-hash-1'),
getBlock: sandbox.stub().resolves({ tx: [], time: 1500000000 })
},
filterBlock: {
filterMemoTxs: sandbox.stub().resolves([])
}
}
uut = new IndexBlocks({ adapters })
uut.retryQueue = {
addToQueue: sandbox.stub().callsFake(async (fn, arg) => {
if (fn === adapters.rpc.getBlockHash) return adapters.rpc.getBlockHash(arg)
if (fn === adapters.rpc.getBlock) return adapters.rpc.getBlock(arg)
throw new Error('Unexpected addToQueue call')
})
}
uut.filterBlock = adapters.filterBlock
})
afterEach(() => sandbox.restore())
describe('#processMemoTxs', () => {
it('should process all memo txs in parallel', async () => {
const processMemoTx = sandbox.stub(uut, 'processMemoTx').resolves(true)
const blockSeen = 1500000000000
await uut.processMemoTxs(['tx-a', 'tx-b', 'tx-c'], 600000, blockSeen)
assert.equal(processMemoTx.callCount, 3)
assert.deepEqual(
processMemoTx.getCalls().map((call) => call.args[0]).sort(),
['tx-a', 'tx-b', 'tx-c']
)
processMemoTx.getCalls().forEach((call) => {
assert.equal(call.args[1], 600000)
assert.equal(call.args[2], blockSeen)
})
})
it('should fail the block if any memo tx fails', async () => {
sandbox.stub(uut, 'processMemoTx')
.onFirstCall().resolves(true)
.onSecondCall().rejects(new Error('db error'))
.onThirdCall().resolves(true)
try {
await uut.processMemoTxs(['tx-a', 'tx-b', 'tx-c'], 600000, 1500000000000)
assert.fail('Expected processMemoTxs to throw')
} catch (err) {
assert.include(err.message, 'db error')
}
})
})
describe('#processBlock', () => {
it('should pass block.time in milliseconds to processMemoTxs', async () => {
const processMemoTxs = sandbox.stub(uut, 'processMemoTxs').resolves(true)
adapters.filterBlock.filterMemoTxs.resolves(['tx-a', 'tx-b'])
await uut.processBlock(600000)
assert.equal(processMemoTxs.callCount, 1)
assert.deepEqual(processMemoTxs.firstCall.args, [['tx-a', 'tx-b'], 600000, 1500000000000])
})
})
})