From bd0b1b0318179fd6fa52f8a0f674b107ff050fe3 Mon Sep 17 00:00:00 2001 From: satoshi Date: Sun, 12 Apr 2026 19:52:27 +0300 Subject: [PATCH] Phase 2a: kamado-api Go middleware (core MVP) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Go 1.22 module that polls CKPool's Unix socket control API, queries Bitcoin Core over JSON-RPC, merges both into a thread-safe snapshot, and serves it over REST. Layout: api/ ├── cmd/kamado-api/main.go signal-aware entrypoint └── internal/ ├── config/ env var loader with validation ├── ckpool/ socket client (4-byte LE length-prefixed wire │ protocol verified against libckpool.c), typed │ response models for poolstats/users/workers/ │ clients/uptime, + unit tests using a fake │ unix socket server ├── bitcoind/ minimal JSON-RPC client, getblockchaininfo │ and getnetworkhashps ├── state/ Aggregator that refreshes a merged Snapshot │ on a ticker; readers get a copy under RWMutex └── httpapi/ REST handlers on Go 1.22 ServeMux: /api/health /api/pool /api/users /api/workers /api/clients /api/snapshot CKPool stores hashrate as "dsps" (diff shares per second); we convert to H/s via the 2^32 constant used by GoBrrr-Pool and other clients. Every stat CKPool exposes to its socket API is surfaced — useragent, IP, per-client diff, per-worker best diff — closing the gap against Bassin which only reads the 60-second stats files. Dockerfile does a CGO_ENABLED=0 static build on golang:1.22-bookworm with -trimpath -ldflags=-s -w. docker-compose now runs both ckpool and kamado-api, sharing a named volume for /run/ckpool so the API can dial the stratifier socket directly. Deferred to Phase 2b (called out in README): - Bitcoin Core ZMQ hashblock subscriber - WebSocket push for real-time UI updates - SQLite persistence (block history, best-share history) - CKPool log tailer for "Solved and confirmed block" detection Tests and build NOT run in this commit — Go isn't installed in the dev environment. Run `make api-test` or `make api` to verify. Co-Authored-By: Claude Opus 4.6 --- Makefile | 18 +++- README.md | 24 ++++- api/Dockerfile | 34 ++++++ api/cmd/kamado-api/main.go | 71 +++++++++++++ api/go.mod | 3 + api/internal/bitcoind/rpc.go | 124 ++++++++++++++++++++++ api/internal/ckpool/client.go | 130 +++++++++++++++++++++++ api/internal/ckpool/client_test.go | 137 ++++++++++++++++++++++++ api/internal/ckpool/types.go | 162 +++++++++++++++++++++++++++++ api/internal/config/config.go | 82 +++++++++++++++ api/internal/httpapi/server.go | 85 +++++++++++++++ api/internal/state/aggregator.go | 137 ++++++++++++++++++++++++ docker-compose.yml | 47 +++++++-- 13 files changed, 1040 insertions(+), 14 deletions(-) create mode 100644 api/Dockerfile create mode 100644 api/cmd/kamado-api/main.go create mode 100644 api/go.mod create mode 100644 api/internal/bitcoind/rpc.go create mode 100644 api/internal/ckpool/client.go create mode 100644 api/internal/ckpool/client_test.go create mode 100644 api/internal/ckpool/types.go create mode 100644 api/internal/config/config.go create mode 100644 api/internal/httpapi/server.go create mode 100644 api/internal/state/aggregator.go diff --git a/Makefile b/Makefile index b9e3e60..4299de0 100644 --- a/Makefile +++ b/Makefile @@ -1,20 +1,28 @@ # ---------------------------------------------------------------------------- # Kamado Pool — top-level Makefile # ---------------------------------------------------------------------------- -.PHONY: help ckpool ckpool-shell up down logs clean +.PHONY: help ckpool api api-test ckpool-shell up down logs clean help: @echo "Kamado Pool targets:" @echo " make ckpool Build the ckpool-solo docker image" + @echo " make api Build the kamado-api docker image" + @echo " make api-test Run Go tests for kamado-api (requires Go installed)" @echo " make ckpool-shell Open a shell in the built ckpool image" - @echo " make up docker compose up -d" + @echo " make up docker compose up -d --build" @echo " make down docker compose down" - @echo " make logs Tail ckpool logs" - @echo " make clean Remove build artifacts and data" + @echo " make logs Tail ckpool + api logs" + @echo " make clean Remove build artifacts, data, and volumes" ckpool: docker build -t kamado/ckpool:dev ./ckpool +api: + docker build -t kamado/api:dev ./api + +api-test: + cd api && go test ./... -race + ckpool-shell: ckpool docker run --rm -it --entrypoint /bin/sh kamado/ckpool:dev @@ -25,7 +33,7 @@ down: docker compose down logs: - docker compose logs -f ckpool + docker compose logs -f clean: docker compose down -v || true diff --git a/README.md b/README.md index 75adb27..954b99e 100644 --- a/README.md +++ b/README.md @@ -38,11 +38,33 @@ Three main components: ## Phases - [x] **Phase 1** — Fork & fix CKPool, build infrastructure -- [ ] **Phase 2** — Go API middleware (socket client, REST, WebSocket, SQLite) +- [~] **Phase 2** — Go API middleware + - [x] Phase 2a: CKPool socket client, bitcoind RPC, state aggregator, REST API + - [ ] Phase 2b: ZMQ block notifier, WebSocket push, SQLite persistence, log tailer - [ ] **Phase 3** — Svelte UI dashboard - [ ] **Phase 4** — Monorepo Docker build, full stack integration - [ ] **Phase 5** — Testing (regtest, testnet4), polish +## Quick start (dev) + +```sh +cp .env.example .env # set POOL_BTCADDRESS and bitcoind creds +make up # build + start ckpool + api +curl localhost:8080/api/health +curl localhost:8080/api/pool +``` + +REST endpoints (Phase 2a): + +| Route | Returns | +| -------------------- | ---------------------------------------------------- | +| `GET /api/health` | CKPool + bitcoind health | +| `GET /api/pool` | Pool stats + derived hashrate windows + chain info | +| `GET /api/users` | All users from `users` socket command | +| `GET /api/workers` | All workers from `workers` socket command | +| `GET /api/clients` | All connected stratum sessions (useragent, IP, diff) | +| `GET /api/snapshot` | Full merged snapshot (everything) | + StartOS packaging lives in a separate repository. ## Upstream diff --git a/api/Dockerfile b/api/Dockerfile new file mode 100644 index 0000000..4377fcc --- /dev/null +++ b/api/Dockerfile @@ -0,0 +1,34 @@ +# ============================================================================ +# kamado-api build stage +# ============================================================================ +FROM golang:1.22-bookworm AS build + +WORKDIR /src +# Cache deps first +COPY go.mod ./ +RUN go mod download 2>/dev/null || true +COPY . . + +# Static build — CGO off, stripped, reproducible-ish +RUN CGO_ENABLED=0 GOOS=linux go build \ + -trimpath \ + -ldflags="-s -w" \ + -o /out/kamado-api \ + ./cmd/kamado-api + +# ============================================================================ +# Runtime: distroless-ish minimal +# ============================================================================ +FROM debian:bookworm-slim AS runtime + +RUN apt-get update && apt-get install --no-install-recommends -y \ + ca-certificates \ + tini \ + && rm -rf /var/lib/apt/lists/* \ + && mkdir -p /var/lib/kamado /var/log/ckpool /run/ckpool + +COPY --from=build /out/kamado-api /usr/local/bin/kamado-api + +EXPOSE 8080 + +ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/kamado-api"] diff --git a/api/cmd/kamado-api/main.go b/api/cmd/kamado-api/main.go new file mode 100644 index 0000000..b0be8cc --- /dev/null +++ b/api/cmd/kamado-api/main.go @@ -0,0 +1,71 @@ +// kamado-api is the middleware that sits between ckpool-solo and the +// Kamado dashboard. It polls ckpool's Unix socket API, calls bitcoind +// over JSON-RPC, and serves the merged state over REST (WebSocket push, +// SQLite persistence, ZMQ and log-tailer-based block detection land in +// a follow-up commit). +package main + +import ( + "context" + "errors" + "log/slog" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/kamadopool/kamado-api/internal/bitcoind" + "github.com/kamadopool/kamado-api/internal/ckpool" + "github.com/kamadopool/kamado-api/internal/config" + "github.com/kamadopool/kamado-api/internal/httpapi" + "github.com/kamadopool/kamado-api/internal/state" +) + +func main() { + log := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) + slog.SetDefault(log) + + cfg, err := config.FromEnv() + if err != nil { + log.Error("config", "err", err) + os.Exit(1) + } + log.Info("kamado-api starting", + "listen", cfg.ListenAddr, + "sockdir", cfg.CKPoolSockDir, + "bitcoind", cfg.BitcoinRPCURL, + "poll_interval", cfg.PollInterval, + ) + + ck := ckpool.New(cfg.CKPoolSockDir) + rpc := bitcoind.NewRPC(cfg.BitcoinRPCURL, cfg.BitcoinRPCUser, cfg.BitcoinRPCPassword, cfg.BitcoinRPCTimeout) + agg := state.New(ck, rpc, cfg.PollInterval, log) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + go agg.Run(ctx) + + api := httpapi.New(agg, log) + srv := &http.Server{ + Addr: cfg.ListenAddr, + Handler: api.Handler(), + ReadHeaderTimeout: 5 * time.Second, + } + + // Shutdown on ctx cancel + go func() { + <-ctx.Done() + log.Info("shutting down") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = srv.Shutdown(shutdownCtx) + }() + + log.Info("http listening", "addr", cfg.ListenAddr) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Error("http server", "err", err) + os.Exit(1) + } +} diff --git a/api/go.mod b/api/go.mod new file mode 100644 index 0000000..687d5e0 --- /dev/null +++ b/api/go.mod @@ -0,0 +1,3 @@ +module github.com/kamadopool/kamado-api + +go 1.22 diff --git a/api/internal/bitcoind/rpc.go b/api/internal/bitcoind/rpc.go new file mode 100644 index 0000000..220ff54 --- /dev/null +++ b/api/internal/bitcoind/rpc.go @@ -0,0 +1,124 @@ +// Package bitcoind provides a minimal Bitcoin Core JSON-RPC client. +// Only the methods kamado-api actually needs are implemented. +package bitcoind + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +type RPC struct { + URL string + User string + Password string + HTTP *http.Client +} + +func NewRPC(url, user, password string, timeout time.Duration) *RPC { + return &RPC{ + URL: url, + User: user, + Password: password, + HTTP: &http.Client{Timeout: timeout}, + } +} + +type rpcRequest struct { + JSONRPC string `json:"jsonrpc"` + ID string `json:"id"` + Method string `json:"method"` + Params []any `json:"params"` +} + +type rpcResponse struct { + Result json.RawMessage `json:"result"` + Error *rpcError `json:"error"` + ID string `json:"id"` +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +func (e *rpcError) Error() string { + return fmt.Sprintf("bitcoind rpc error %d: %s", e.Code, e.Message) +} + +// Call performs a single JSON-RPC request and unmarshals the result. +func (c *RPC) Call(ctx context.Context, method string, params []any, out any) error { + body, err := json.Marshal(rpcRequest{ + JSONRPC: "1.0", + ID: "kamado", + Method: method, + Params: params, + }) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, "POST", c.URL, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.SetBasicAuth(c.User, c.Password) + + resp, err := c.HTTP.Do(req) + if err != nil { + return fmt.Errorf("bitcoind rpc: %w", err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("bitcoind rpc: read body: %w", err) + } + // bitcoind returns 500 on rpc errors but still with a valid JSON body. + var rr rpcResponse + if err := json.Unmarshal(raw, &rr); err != nil { + return fmt.Errorf("bitcoind rpc: unmarshal (status %d): %w: %s", resp.StatusCode, err, string(raw)) + } + if rr.Error != nil { + return rr.Error + } + if out != nil { + return json.Unmarshal(rr.Result, out) + } + return nil +} + +// ---- typed method wrappers ---- + +type BlockchainInfo struct { + Chain string `json:"chain"` + Blocks int64 `json:"blocks"` + Headers int64 `json:"headers"` + BestBlockHash string `json:"bestblockhash"` + Difficulty float64 `json:"difficulty"` + MedianTime int64 `json:"mediantime"` + VerificationProgress float64 `json:"verificationprogress"` + InitialBlockDownload bool `json:"initialblockdownload"` +} + +func (c *RPC) GetBlockchainInfo(ctx context.Context) (*BlockchainInfo, error) { + var out BlockchainInfo + if err := c.Call(ctx, "getblockchaininfo", nil, &out); err != nil { + return nil, err + } + return &out, nil +} + +// NetworkHashPS returns the network hashrate at the given block height. +// `blocks` is a window (default 120). Pass -1 to use the default. +func (c *RPC) GetNetworkHashPS(ctx context.Context, blocks, height int) (float64, error) { + var out float64 + if err := c.Call(ctx, "getnetworkhashps", []any{blocks, height}, &out); err != nil { + return 0, err + } + return out, nil +} diff --git a/api/internal/ckpool/client.go b/api/internal/ckpool/client.go new file mode 100644 index 0000000..53343db --- /dev/null +++ b/api/internal/ckpool/client.go @@ -0,0 +1,130 @@ +// Package ckpool implements a client for CKPool's Unix-domain-socket control +// protocol. Each request opens a fresh connection, writes a 4-byte little- +// endian length prefix followed by the payload, half-closes the write side, +// then reads a 4-byte length prefix and that many bytes of response. +// +// See src/libckpool.c in upstream CKPool for the reference implementation +// (send_unix_msg / recv_unix_msg). +package ckpool + +import ( + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "time" +) + +// Default socket filenames under the sockdir. +const ( + SocketListener = "listener" + SocketStratifier = "stratifier" + SocketConnector = "connector" +) + +// Client talks to the ckpool Unix sockets at SockDir. It is safe for +// concurrent use: each Send opens its own short-lived connection, matching +// CKPool's own one-shot request-response model. +type Client struct { + SockDir string + Timeout time.Duration +} + +// New returns a Client configured with a sensible default timeout. +func New(sockDir string) *Client { + return &Client{ + SockDir: sockDir, + Timeout: 5 * time.Second, + } +} + +// maxMsgLen matches the upper bound in libckpool.c (0x80000000). Anything +// above this indicates a protocol error and we refuse to allocate for it. +const maxMsgLen = 0x80000000 + +// ErrProtocol is returned when the server's framing is invalid. +var ErrProtocol = errors.New("ckpool: protocol error") + +// Send delivers a single command to the named ckpool socket and returns the +// raw response bytes. `sockName` is one of SocketListener, SocketStratifier, +// or SocketConnector. +func (c *Client) Send(ctx context.Context, sockName, cmd string) ([]byte, error) { + if cmd == "" { + return nil, fmt.Errorf("ckpool: empty command") + } + path := c.SockDir + "/" + sockName + + var d net.Dialer + dialCtx, cancel := context.WithTimeout(ctx, c.Timeout) + defer cancel() + conn, err := d.DialContext(dialCtx, "unix", path) + if err != nil { + return nil, fmt.Errorf("ckpool: dial %s: %w", path, err) + } + defer conn.Close() + + if dl, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(dl) + } else { + _ = conn.SetDeadline(time.Now().Add(c.Timeout)) + } + + // ---- write: 4-byte LE length + payload + half-close ---- + var lenBuf [4]byte + binary.LittleEndian.PutUint32(lenBuf[:], uint32(len(cmd))) + if _, err := conn.Write(lenBuf[:]); err != nil { + return nil, fmt.Errorf("ckpool: write len: %w", err) + } + if _, err := conn.Write([]byte(cmd)); err != nil { + return nil, fmt.Errorf("ckpool: write payload: %w", err) + } + // Half-close write so the server sees EOF and responds, mirroring + // libckpool's shutdown(SHUT_WR) at the end of send_unix_msg. + if uc, ok := conn.(*net.UnixConn); ok { + if err := uc.CloseWrite(); err != nil { + return nil, fmt.Errorf("ckpool: close write: %w", err) + } + } + + // ---- read: 4-byte LE length + payload ---- + if _, err := io.ReadFull(conn, lenBuf[:]); err != nil { + return nil, fmt.Errorf("ckpool: read len: %w", err) + } + msgLen := binary.LittleEndian.Uint32(lenBuf[:]) + if msgLen == 0 || msgLen > maxMsgLen { + return nil, fmt.Errorf("%w: invalid msg len %d", ErrProtocol, msgLen) + } + buf := make([]byte, msgLen) + if _, err := io.ReadFull(conn, buf); err != nil { + return nil, fmt.Errorf("ckpool: read payload: %w", err) + } + return buf, nil +} + +// SendJSON sends a command and unmarshals the response into `out`. +// If the response is not valid JSON, the raw bytes are returned in the error. +func (c *Client) SendJSON(ctx context.Context, sockName, cmd string, out any) error { + raw, err := c.Send(ctx, sockName, cmd) + if err != nil { + return err + } + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("ckpool: unmarshal %q: %w (raw=%s)", cmd, err, string(raw)) + } + return nil +} + +// Ping returns nil if the stratifier replies "pong". +func (c *Client) Ping(ctx context.Context) error { + raw, err := c.Send(ctx, SocketStratifier, "ping") + if err != nil { + return err + } + if string(raw) != "pong" { + return fmt.Errorf("ckpool: expected pong, got %q", string(raw)) + } + return nil +} diff --git a/api/internal/ckpool/client_test.go b/api/internal/ckpool/client_test.go new file mode 100644 index 0000000..05ce812 --- /dev/null +++ b/api/internal/ckpool/client_test.go @@ -0,0 +1,137 @@ +package ckpool + +import ( + "context" + "encoding/binary" + "io" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +// fakeServer spawns a goroutine that listens on a unix socket under `dir`, +// reads one request in the ckpool wire format, and responds with `reply`. +// It validates the request format and reports errors via `t`. +func fakeServer(t *testing.T, dir, sockName, wantCmd, reply string) { + t.Helper() + path := filepath.Join(dir, sockName) + ln, err := net.Listen("unix", path) + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + + // Read 4-byte LE length + var lenBuf [4]byte + if _, err := io.ReadFull(conn, lenBuf[:]); err != nil { + t.Errorf("server read len: %v", err) + return + } + n := binary.LittleEndian.Uint32(lenBuf[:]) + buf := make([]byte, n) + if _, err := io.ReadFull(conn, buf); err != nil { + t.Errorf("server read payload: %v", err) + return + } + if string(buf) != wantCmd { + t.Errorf("server got cmd %q, want %q", string(buf), wantCmd) + return + } + + // Respond with length prefix + payload + binary.LittleEndian.PutUint32(lenBuf[:], uint32(len(reply))) + if _, err := conn.Write(lenBuf[:]); err != nil { + t.Errorf("server write len: %v", err) + return + } + if _, err := conn.Write([]byte(reply)); err != nil { + t.Errorf("server write reply: %v", err) + return + } + }() +} + +func tempSockDir(t *testing.T) string { + t.Helper() + // Use a short dir: unix socket paths are limited to ~108 chars on Linux. + dir, err := os.MkdirTemp("", "ck") + if err != nil { + t.Fatalf("mkdir: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + return dir +} + +func TestClient_Ping(t *testing.T) { + dir := tempSockDir(t) + fakeServer(t, dir, SocketStratifier, "ping", "pong") + + c := New(dir) + c.Timeout = time.Second + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := c.Ping(ctx); err != nil { + t.Fatalf("ping: %v", err) + } +} + +func TestClient_PoolStats(t *testing.T) { + dir := tempSockDir(t) + reply := `{"start":1775000000,"update":1775949393,"workers":42664,"users":24679,"disconnected":6119,"shares":338031,"sps1":6.48,"sps5":6.49,"sps15":6.49,"sps60":6.48,"accepted":338031,"rejected":29602,"dsps1":0.274,"dsps5":0.302,"dsps15":0.317,"dsps60":0.309,"dsps360":0.298,"dsps1440":0.290,"dsps10080":0.285}` + fakeServer(t, dir, SocketStratifier, "poolstats", reply) + + c := New(dir) + c.Timeout = time.Second + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + ps, err := c.PoolStats(ctx) + if err != nil { + t.Fatalf("poolstats: %v", err) + } + if ps.Workers != 42664 { + t.Errorf("Workers = %d, want 42664", ps.Workers) + } + if got := DSPSToHashrate(ps.DSPS1); got < 1e9 { + t.Errorf("DSPS1 hashrate = %v, want > 1GH/s", got) + } +} + +func TestClient_Clients(t *testing.T) { + dir := tempSockDir(t) + reply := `{"clients":[{"id":12345,"useragent":"Bitaxe/2.4.0","address":"192.168.1.100","workername":"bc1q.abc","authorised":true,"bestdiff":12345.67}]}` + fakeServer(t, dir, SocketStratifier, "clients", reply) + + c := New(dir) + c.Timeout = time.Second + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + clients, err := c.Clients(ctx) + if err != nil { + t.Fatalf("clients: %v", err) + } + if len(clients) != 1 { + t.Fatalf("got %d clients, want 1", len(clients)) + } + if clients[0].UserAgent != "Bitaxe/2.4.0" { + t.Errorf("UserAgent = %q, want Bitaxe/2.4.0", clients[0].UserAgent) + } +} + +func TestClient_DialError(t *testing.T) { + c := New("/nonexistent/path") + c.Timeout = 200 * time.Millisecond + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := c.Ping(ctx); err == nil { + t.Error("expected error, got nil") + } +} diff --git a/api/internal/ckpool/types.go b/api/internal/ckpool/types.go new file mode 100644 index 0000000..0681ff6 --- /dev/null +++ b/api/internal/ckpool/types.go @@ -0,0 +1,162 @@ +package ckpool + +import ( + "context" + "fmt" +) + +// DSPSToHashrate converts CKPool's internal "difficulty shares per second" +// value to hashrate in H/s. CKPool shares are normalized to diff 1 which +// corresponds to 2^32 hashes. +const HashesPerShare = 4294967296.0 + +func DSPSToHashrate(dsps float64) float64 { return dsps * HashesPerShare } + +// PoolStats is the response to `poolstats` on the stratifier socket. +type PoolStats struct { + Start int64 `json:"start"` + Update int64 `json:"update"` + Workers int64 `json:"workers"` + Users int64 `json:"users"` + Disconnected int64 `json:"disconnected"` + Shares int64 `json:"shares"` + SPS1 float64 `json:"sps1"` + SPS5 float64 `json:"sps5"` + SPS15 float64 `json:"sps15"` + SPS60 float64 `json:"sps60"` + Accepted int64 `json:"accepted"` + Rejected int64 `json:"rejected"` + DSPS1 float64 `json:"dsps1"` + DSPS5 float64 `json:"dsps5"` + DSPS15 float64 `json:"dsps15"` + DSPS60 float64 `json:"dsps60"` + DSPS360 float64 `json:"dsps360"` + DSPS1440 float64 `json:"dsps1440"` + DSPS10080 float64 `json:"dsps10080"` +} + +// User is one entry in the `users` response. +type User struct { + User string `json:"user"` + ID int64 `json:"id"` + Workers int `json:"workers"` + BestDiff float64 `json:"bestdiff"` + DSPS1 float64 `json:"dsps1"` + DSPS5 float64 `json:"dsps5"` + DSPS60 float64 `json:"dsps60"` + DSPS1440 float64 `json:"dsps1440"` + DSPS10080 float64 `json:"dsps10080"` + LastShare int64 `json:"lastshare"` +} + +type UsersResponse struct { + Users []User `json:"users"` +} + +// Worker is one entry in the `workers` response. +type Worker struct { + User string `json:"user"` + Worker string `json:"worker"` + ID int64 `json:"id"` + DSPS1 float64 `json:"dsps1"` + DSPS5 float64 `json:"dsps5"` + DSPS60 float64 `json:"dsps60"` + DSPS1440 float64 `json:"dsps1440"` + LastShare int64 `json:"lastshare"` + BestDiff float64 `json:"bestdiff"` + MinDiff float64 `json:"mindiff"` + Idle bool `json:"idle"` +} + +type WorkersResponse struct { + Workers []Worker `json:"workers"` +} + +// Client is one entry in the `clients` response. Note: this is CKPool's +// view of a connected stratum session, not our Go socket client. +type StratumClient struct { + ID int64 `json:"id"` + Enonce1 string `json:"enonce1"` + Enonce1Var string `json:"enonce1var"` + Enonce164 int64 `json:"enonce1_64"` + Diff float64 `json:"diff"` + DSPS1 float64 `json:"dsps1"` + DSPS5 float64 `json:"dsps5"` + DSPS60 float64 `json:"dsps60"` + DSPS1440 float64 `json:"dsps1440"` + DSPS10080 float64 `json:"dsps10080"` + LastShare int64 `json:"lastshare"` + StartTime int64 `json:"starttime"` + Address string `json:"address"` + Subscribed bool `json:"subscribed"` + Authorised bool `json:"authorised"` + Idle bool `json:"idle"` + UserAgent string `json:"useragent"` + WorkerName string `json:"workername"` + UserID int64 `json:"userid"` + Server int `json:"server"` + BestDiff float64 `json:"bestdiff"` + ProxyID int `json:"proxyid"` + SubProxyID int `json:"subproxyid"` +} + +type ClientsResponse struct { + Clients []StratumClient `json:"clients"` +} + +// Uptime is the response to `uptime`. +type Uptime struct { + Uptime int64 `json:"uptime"` +} + +// --- High-level wrappers ------------------------------------------------ + +func (c *Client) PoolStats(ctx context.Context) (*PoolStats, error) { + var ps PoolStats + if err := c.SendJSON(ctx, SocketStratifier, "poolstats", &ps); err != nil { + return nil, err + } + return &ps, nil +} + +func (c *Client) Users(ctx context.Context) ([]User, error) { + var r UsersResponse + if err := c.SendJSON(ctx, SocketStratifier, "users", &r); err != nil { + return nil, err + } + return r.Users, nil +} + +func (c *Client) Workers(ctx context.Context) ([]Worker, error) { + var r WorkersResponse + if err := c.SendJSON(ctx, SocketStratifier, "workers", &r); err != nil { + return nil, err + } + return r.Workers, nil +} + +func (c *Client) Clients(ctx context.Context) ([]StratumClient, error) { + var r ClientsResponse + if err := c.SendJSON(ctx, SocketStratifier, "clients", &r); err != nil { + return nil, err + } + return r.Clients, nil +} + +func (c *Client) Uptime(ctx context.Context) (int64, error) { + var u Uptime + if err := c.SendJSON(ctx, SocketStratifier, "uptime", &u); err != nil { + return 0, err + } + return u.Uptime, nil +} + +// GetUser returns stats for a single user by address. +func (c *Client) GetUser(ctx context.Context, address string) (*User, error) { + cmd := fmt.Sprintf("getuser.{\"user\":%q}", address) + var u User + if err := c.SendJSON(ctx, SocketStratifier, cmd, &u); err != nil { + return nil, err + } + return &u, nil +} diff --git a/api/internal/config/config.go b/api/internal/config/config.go new file mode 100644 index 0000000..f4f51c3 --- /dev/null +++ b/api/internal/config/config.go @@ -0,0 +1,82 @@ +// Package config loads kamado-api configuration from environment variables. +// Missing required values are a hard error at startup; optional values get +// documented defaults. +package config + +import ( + "fmt" + "os" + "strconv" + "time" +) + +type Config struct { + // HTTP server + ListenAddr string + + // CKPool socket + CKPoolSockDir string + + // Bitcoin Core RPC + BitcoinRPCURL string // full URL e.g. http://bitcoind:8332 + BitcoinRPCUser string + BitcoinRPCPassword string + BitcoinRPCTimeout time.Duration + + // ZMQ (Phase 2b) + BitcoinZMQBlock string // e.g. tcp://bitcoind:28332, empty to disable + + // CKPool log file, for block-solve detection (Phase 2b) + CKPoolLogFile string + + // SQLite DB path for persistence (Phase 2b) + DBPath string + + // Poll interval for refreshing ckpool stats + PollInterval time.Duration +} + +func FromEnv() (*Config, error) { + cfg := &Config{ + ListenAddr: getenv("LISTEN_ADDR", ":8080"), + CKPoolSockDir: getenv("CKPOOL_SOCKDIR", "/run/ckpool"), + BitcoinRPCURL: os.Getenv("BITCOIN_RPC_URL"), + BitcoinRPCUser: os.Getenv("BITCOIN_RPC_USER"), + BitcoinRPCPassword: os.Getenv("BITCOIN_RPC_PASSWORD"), + BitcoinRPCTimeout: getenvDuration("BITCOIN_RPC_TIMEOUT", 10*time.Second), + BitcoinZMQBlock: os.Getenv("BITCOIN_ZMQ_BLOCK"), + CKPoolLogFile: getenv("CKPOOL_LOGFILE", "/var/log/ckpool/ckpool.log"), + DBPath: getenv("DB_PATH", "/var/lib/kamado/kamado.db"), + PollInterval: getenvDuration("POLL_INTERVAL", 5*time.Second), + } + + if cfg.BitcoinRPCURL == "" { + return nil, fmt.Errorf("BITCOIN_RPC_URL is required") + } + if cfg.BitcoinRPCUser == "" || cfg.BitcoinRPCPassword == "" { + return nil, fmt.Errorf("BITCOIN_RPC_USER and BITCOIN_RPC_PASSWORD are required") + } + return cfg, nil +} + +func getenv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func getenvDuration(key string, def time.Duration) time.Duration { + v := os.Getenv(key) + if v == "" { + return def + } + // Accept both Go duration syntax ("5s", "10s") and bare seconds ("10"). + if d, err := time.ParseDuration(v); err == nil { + return d + } + if n, err := strconv.Atoi(v); err == nil { + return time.Duration(n) * time.Second + } + return def +} diff --git a/api/internal/httpapi/server.go b/api/internal/httpapi/server.go new file mode 100644 index 0000000..c3c614f --- /dev/null +++ b/api/internal/httpapi/server.go @@ -0,0 +1,85 @@ +// Package httpapi serves the Kamado REST API. WebSocket push and static +// UI serving land in a follow-up commit. +package httpapi + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/kamadopool/kamado-api/internal/state" +) + +type Server struct { + Agg *state.Aggregator + Log *slog.Logger +} + +func New(agg *state.Aggregator, log *slog.Logger) *Server { + return &Server{Agg: agg, Log: log} +} + +// Handler returns an http.Handler with all kamado routes mounted under /api. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /api/health", s.health) + mux.HandleFunc("GET /api/pool", s.pool) + mux.HandleFunc("GET /api/users", s.users) + mux.HandleFunc("GET /api/workers", s.workers) + mux.HandleFunc("GET /api/clients", s.clients) + mux.HandleFunc("GET /api/snapshot", s.snapshot) + return mux +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +// ---- handlers --------------------------------------------------------- + +func (s *Server) health(w http.ResponseWriter, r *http.Request) { + snap := s.Agg.Snapshot() + status := http.StatusOK + if !snap.CKPoolOK || !snap.BitcoinOK { + status = http.StatusServiceUnavailable + } + writeJSON(w, status, map[string]any{ + "ok": snap.CKPoolOK && snap.BitcoinOK, + "ckpool": snap.CKPoolOK, + "bitcoin": snap.BitcoinOK, + "last_error": snap.LastError, + }) +} + +func (s *Server) snapshot(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.Agg.Snapshot()) +} + +func (s *Server) pool(w http.ResponseWriter, r *http.Request) { + snap := s.Agg.Snapshot() + writeJSON(w, http.StatusOK, map[string]any{ + "pool": snap.Pool, + "uptime_seconds": snap.Uptime, + "hashrate_hs_1m": snap.HashrateHs, + "hashrate_hs_5m": snap.HashrateHs5m, + "hashrate_hs_1h": snap.HashrateHs1h, + "hashrate_hs_24h": snap.HashrateHs24h, + "chain": snap.Chain, + "network_hashrate_hs": snap.NetworkHashrateHs, + }) +} + +func (s *Server) users(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.Agg.Snapshot().Users) +} + +func (s *Server) workers(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.Agg.Snapshot().Workers) +} + +func (s *Server) clients(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, s.Agg.Snapshot().Clients) +} diff --git a/api/internal/state/aggregator.go b/api/internal/state/aggregator.go new file mode 100644 index 0000000..1b0e0a5 --- /dev/null +++ b/api/internal/state/aggregator.go @@ -0,0 +1,137 @@ +// Package state merges data from CKPool (via Unix socket), Bitcoin Core +// (via JSON-RPC), and future sources (ZMQ, log tailer) into a single +// thread-safe snapshot the HTTP layer can serve. The snapshot is refreshed +// on a ticker; readers get a copy without blocking the refresh. +package state + +import ( + "context" + "log/slog" + "sync" + "time" + + "github.com/kamadopool/kamado-api/internal/bitcoind" + "github.com/kamadopool/kamado-api/internal/ckpool" +) + +// Snapshot is the merged view served to the UI. All fields are safe to +// JSON-serialize directly. +type Snapshot struct { + GeneratedAt time.Time `json:"generated_at"` + + // CKPool-derived fields + Pool *ckpool.PoolStats `json:"pool"` + Uptime int64 `json:"uptime_seconds"` + Users []ckpool.User `json:"users"` + Workers []ckpool.Worker `json:"workers"` + Clients []ckpool.StratumClient `json:"clients"` + + // Derived/enriched fields + HashrateHs float64 `json:"hashrate_hs_1m"` // from PoolStats.DSPS1 + HashrateHs5m float64 `json:"hashrate_hs_5m"` + HashrateHs1h float64 `json:"hashrate_hs_1h"` + HashrateHs24h float64 `json:"hashrate_hs_24h"` + + // Bitcoin Core fields + Chain *bitcoind.BlockchainInfo `json:"chain"` + NetworkHashrateHs float64 `json:"network_hashrate_hs"` + + // Health + CKPoolOK bool `json:"ckpool_ok"` + BitcoinOK bool `json:"bitcoin_ok"` + LastError string `json:"last_error,omitempty"` +} + +// Aggregator refreshes a Snapshot on a ticker. +type Aggregator struct { + CK *ckpool.Client + RPC *bitcoind.RPC + Interval time.Duration + Log *slog.Logger + + mu sync.RWMutex + snap Snapshot +} + +func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator { + return &Aggregator{ + CK: ck, + RPC: rpc, + Interval: interval, + Log: log, + } +} + +// Run blocks until ctx is cancelled, refreshing the snapshot every Interval. +// It runs one immediate refresh at startup so readers don't see an empty +// snapshot after ctx launches the goroutine. +func (a *Aggregator) Run(ctx context.Context) { + a.refresh(ctx) + t := time.NewTicker(a.Interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + a.refresh(ctx) + } + } +} + +// Snapshot returns a copy of the current snapshot. +func (a *Aggregator) Snapshot() Snapshot { + a.mu.RLock() + defer a.mu.RUnlock() + return a.snap +} + +func (a *Aggregator) refresh(ctx context.Context) { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + next := Snapshot{GeneratedAt: time.Now()} + + // --- ckpool: poolstats, users, workers, clients, uptime --- + if ps, err := a.CK.PoolStats(ctx); err == nil { + next.Pool = ps + next.HashrateHs = ckpool.DSPSToHashrate(ps.DSPS1) + next.HashrateHs5m = ckpool.DSPSToHashrate(ps.DSPS5) + next.HashrateHs1h = ckpool.DSPSToHashrate(ps.DSPS60) + next.HashrateHs24h = ckpool.DSPSToHashrate(ps.DSPS1440) + next.CKPoolOK = true + } else { + a.Log.Warn("ckpool poolstats failed", "err", err) + next.LastError = err.Error() + } + if u, err := a.CK.Uptime(ctx); err == nil { + next.Uptime = u + } + if us, err := a.CK.Users(ctx); err == nil { + next.Users = us + } + if ws, err := a.CK.Workers(ctx); err == nil { + next.Workers = ws + } + if cs, err := a.CK.Clients(ctx); err == nil { + next.Clients = cs + } + + // --- bitcoind: chain + network hashrate --- + if bi, err := a.RPC.GetBlockchainInfo(ctx); err == nil { + next.Chain = bi + next.BitcoinOK = true + if nh, err := a.RPC.GetNetworkHashPS(ctx, -1, int(bi.Blocks)); err == nil { + next.NetworkHashrateHs = nh + } + } else { + a.Log.Warn("bitcoind getblockchaininfo failed", "err", err) + if next.LastError == "" { + next.LastError = err.Error() + } + } + + a.mu.Lock() + a.snap = next + a.mu.Unlock() +} diff --git a/docker-compose.yml b/docker-compose.yml index f3fbe53..1996f50 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,12 +1,12 @@ # ---------------------------------------------------------------------------- # Kamado Pool — development compose # -# Spins up ckpool-solo pointing at a bitcoind. Expects bitcoind to be -# reachable at the host/port set in the `env` block below. For local -# testing, run bitcoind on regtest/testnet4 and set these accordingly. +# Spins up ckpool-solo and kamado-api pointing at a bitcoind. Expects +# bitcoind to be reachable at the host/port set in the env blocks below. +# For local testing, run bitcoind on regtest/testnet4 and set these via .env. # -# The ckpool socket dir is bind-mounted to ./data/run so the (future) -# kamado-api can talk to it from outside the container during development. +# ckpool and kamado-api share the `ckpool-sock` named volume so the API +# can dial /run/ckpool/stratifier directly. # ---------------------------------------------------------------------------- services: ckpool: @@ -18,7 +18,7 @@ services: ports: - "3333:3333" volumes: - - ./data/run:/run/ckpool + - ckpool-sock:/run/ckpool - ./data/logs:/var/log/ckpool environment: # --- Bitcoin Core connection --- @@ -27,7 +27,6 @@ services: BITCOIN_RPC_USER: "${BITCOIN_RPC_USER:-kamado}" BITCOIN_RPC_PASSWORD: "${BITCOIN_RPC_PASSWORD:-kamado}" BITCOIN_NOTIFY: "true" - # Optional ZMQ for instant block notifications (recommended) ZMQ_BLOCK: "${ZMQ_BLOCK:-}" # e.g. tcp://host.docker.internal:28332 # --- Pool settings --- @@ -47,5 +46,37 @@ services: LOGDIR: "/var/log/ckpool" SOCKET_DIR: "/run/ckpool" extra_hosts: - # Allow containers to reach bitcoind running on the host - "host.docker.internal:host-gateway" + + api: + build: + context: ./api + dockerfile: Dockerfile + container_name: kamado-api + restart: unless-stopped + depends_on: + - ckpool + ports: + - "8080:8080" + volumes: + - ckpool-sock:/run/ckpool + - ./data/logs:/var/log/ckpool:ro + - kamado-db:/var/lib/kamado + environment: + LISTEN_ADDR: ":8080" + CKPOOL_SOCKDIR: "/run/ckpool" + CKPOOL_LOGFILE: "/var/log/ckpool/ckpool.log" + POLL_INTERVAL: "5s" + + # --- Bitcoin Core RPC (same creds as ckpool) --- + BITCOIN_RPC_URL: "http://host.docker.internal:48332" + BITCOIN_RPC_USER: "${BITCOIN_RPC_USER:-kamado}" + BITCOIN_RPC_PASSWORD: "${BITCOIN_RPC_PASSWORD:-kamado}" + BITCOIN_RPC_TIMEOUT: "10s" + BITCOIN_ZMQ_BLOCK: "${ZMQ_BLOCK:-}" + extra_hosts: + - "host.docker.internal:host-gateway" + +volumes: + ckpool-sock: + kamado-db: