feat: Phase 1 RPC protocol infrastructure for daemon architecture (bd-111)
Implemented Unix socket RPC foundation to enable daemon-based concurrent access: New files: - internal/rpc/protocol.go: Request/Response types with 13 operations - internal/rpc/server.go: Unix socket server with storage adapter - internal/rpc/client.go: Client with auto-detection and typed methods - internal/rpc/rpc_test.go: Integration tests Features: - JSON-based protocol over Unix sockets - Adapter pattern for context/actor propagation to storage API - Ping/health checks for daemon detection - All core operations: create, update, close, list, show, ready, stats, deps, labels - Graceful socket cleanup and signal handling - Concurrent request support Tests: 49.3% coverage, all passing Related issues: - bd-110: Daemon architecture epic - bd-111: Phase 1 (completed) - bd-112: Phase 2 (client auto-detection) - bd-113: Phase 3 (daemon command) - bd-114: Phase 4 (atomic operations) Amp-Thread-ID: https://ampcode.com/threads/T-796c62e6-93b6-41c7-9cb5-8acc4a35ba9a Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
Steve Yegge
@@ -10,6 +10,11 @@
|
||||
{"id":"bd-107","title":"Make maxDepth configurable in bd dep tree command","description":"Currently maxDepth is hardcoded to 50 in GetDependencyTree. Add --max-depth flag to bd dep tree command to allow users to control recursion depth. Default should remain 50 for safety, but users with very deep trees or wanting shallow views should be able to configure it.","status":"open","priority":4,"issue_type":"feature","created_at":"2025-10-16T20:46:08.971822-07:00","updated_at":"2025-10-16T21:51:08.894613-07:00"}
|
||||
{"id":"bd-108","title":"Fix renumbering temp ID collision bug","description":"bd renumber --force fails with UNIQUE constraint error when trying to assign temp IDs:\n\nError: failed to rename bd-4 to temp ID: failed to update issue ID: constraint failed: UNIQUE constraint failed: issues.id (1555)\n\nThe temp ID generation logic in renumber.go doesn't guarantee unique IDs. Need to:\n1. Use a temp ID strategy that can't collide (e.g., prefix with 'temp-', use UUIDs, or use high numbers like 999999+)\n2. Verify temp IDs don't exist before using them\n3. Add test case for renumbering with various ID gaps\n\nReproduced when renumbering 107 issues with gaps (IDs 1-344 compacting to 1-107).","status":"closed","priority":1,"issue_type":"bug","created_at":"2025-10-16T21:13:38.519915-07:00","updated_at":"2025-10-16T21:51:08.895252-07:00","closed_at":"2025-10-16T21:19:18.49592-07:00"}
|
||||
{"id":"bd-11","title":"Document or automate JSONL sync workflow for git collaboration","description":"When using beads across multiple machines/environments via git, there's a workflow gap:\n\n1. Machine A: Create issues → stored in .beads/project.db\n2. Machine A: bd export -o .beads/issues.jsonl\n3. Machine A: git add .beads/issues.jsonl \u0026\u0026 git commit \u0026\u0026 git push\n4. Machine B: git pull\n5. Machine B: ??? issues.jsonl exists but project.db is empty/stale\n\nThe missing step is: bd import --db .beads/project.db -i .beads/issues.jsonl\n\nThis needs to be either:\na) Documented clearly in workflow docs\nb) Automated (e.g., git hook, or bd auto-imports if jsonl is newer than db)\nc) Both\n\nReal-world impact: User had Claude Code on GCP VM create vc issues from BOOTSTRAP.md. They were exported to issues.jsonl and committed. But on local machine, vc.db was empty until manual import was run.","status":"closed","priority":1,"issue_type":"task","created_at":"2025-10-16T20:46:08.971822-07:00","updated_at":"2025-10-16T21:51:08.743025-07:00","closed_at":"2025-10-14T02:51:52.199766-07:00"}
|
||||
{"id":"bd-110","title":"Implement daemon architecture for concurrent access","description":"Multiple AI agents running concurrently cause database corruption, git lock contention, and data loss. Implement a daemon-based architecture where bd daemon owns SQLite (single writer) and all bd commands become RPC clients when daemon is running. Batches git operations to prevent index.lock contention. Maintains backward compatibility with graceful fallback to direct mode. See DAEMON_DESIGN.md for full details.","design":"Architecture: Unix socket RPC with JSON payloads. bd commands auto-detect daemon socket, fall back to direct mode if not present. Daemon serializes all SQLite writes and batches git exports every 5 seconds. Per-repo daemon using .beads/bd.sock location.\n\nImplementation phases:\n1. RPC protocol infrastructure (protocol.go, server.go, client.go)\n2. Client auto-detection and fallback\n3. Daemon SQLite ownership and git batching\n4. Atomic operations and transactions","acceptance_criteria":"- 4 concurrent agents can run without errors\n- No UNIQUE constraint failures on ID generation\n- No git index.lock errors \n- SQLite counter stays in sync with actual issues\n- Graceful fallback when daemon not running\n- All existing tests pass\n- Documentation updated","status":"open","priority":0,"issue_type":"epic","created_at":"2025-10-16T21:54:48.794119-07:00","updated_at":"2025-10-16T21:54:48.794119-07:00","dependencies":[{"issue_id":"bd-110","depends_on_id":"bd-111","type":"parent-child","created_at":"2025-10-16T21:54:56.032869-07:00","created_by":"stevey"}]}
|
||||
{"id":"bd-111","title":"Phase 1: Implement RPC protocol infrastructure","description":"Create the foundation for daemon-client communication using Unix sockets and JSON.\n\nNew files to create:\n- internal/rpc/protocol.go - Request/response types, operations enum\n- internal/rpc/server.go - Unix socket server that daemon runs\n- internal/rpc/client.go - Client library for bd commands to use\n\nSocket location: .beads/bd.sock (per-repo)\n\nOperations to support initially: create, update, list, show, close, ready, stats","design":"protocol.go defines:\n- Request struct with Operation string and Args json.RawMessage\n- Response struct with Success bool, Data json.RawMessage, Error string\n- Operation constants for all bd commands\n\nserver.go implements:\n- Unix socket listener on .beads/bd.sock\n- Request handler that dispatches to storage layer\n- Graceful shutdown on signals\n\nclient.go implements:\n- TryConnect() to detect running daemon\n- Execute(operation, args) to send RPC request\n- Connection pooling/reuse for performance","acceptance_criteria":"- internal/rpc package compiles without errors\n- Server can accept and respond to simple ping request\n- Client can connect to socket and receive response\n- Unit tests for protocol serialization/deserialization\n- Socket cleanup on server shutdown","status":"closed","priority":0,"issue_type":"task","created_at":"2025-10-16T21:54:48.83081-07:00","updated_at":"2025-10-16T22:02:40.675096-07:00","closed_at":"2025-10-16T22:02:40.675096-07:00"}
|
||||
{"id":"bd-112","title":"Phase 2: Add client auto-detection in bd commands","description":"Modify all bd commands to detect if daemon is running and route through RPC client if available, otherwise fall back to direct storage access.\n\nChanges needed:\n- Update cmd/bd/main.go to check for daemon socket on startup\n- Wrap storage calls with TryConnect logic\n- Ensure all commands work identically in both modes\n- Add --no-daemon flag to force direct mode\n\nThis maintains backward compatibility while enabling daemon mode.","status":"open","priority":0,"issue_type":"task","created_at":"2025-10-16T22:47:36.185502-07:00","updated_at":"2025-10-16T22:47:36.185502-07:00","dependencies":[{"issue_id":"bd-112","depends_on_id":"bd-110","type":"parent-child","created_at":"2025-10-16T22:47:36.190931-07:00","created_by":"stevey"}]}
|
||||
{"id":"bd-113","title":"Phase 3: Implement daemon command with SQLite ownership","description":"Create 'bd daemon' command that starts the RPC server and owns the SQLite database.\n\nImplementation:\n- Add cmd/bd/daemon.go with start/stop/status subcommands\n- Daemon holds exclusive SQLite connection\n- Integrates git sync loop (batch exports every 5 seconds)\n- PID file management for daemon lifecycle\n- Logging for daemon operations\n\nSocket location: .beads/bd.sock per repository","status":"open","priority":0,"issue_type":"task","created_at":"2025-10-16T22:47:42.86546-07:00","updated_at":"2025-10-16T22:47:42.86546-07:00","dependencies":[{"issue_id":"bd-113","depends_on_id":"bd-110","type":"parent-child","created_at":"2025-10-16T22:47:42.874284-07:00","created_by":"stevey"}]}
|
||||
{"id":"bd-114","title":"Phase 4: Add atomic operations and stress testing","description":"Implement atomic multi-operation support and test under concurrent load.\n\nFeatures:\n- Batch/transaction API for multi-step operations\n- Request timeout and cancellation support\n- Connection pooling optimization\n- Stress tests with 4+ concurrent agents\n- Performance benchmarks vs direct mode\n- Documentation updates\n\nValidates all acceptance criteria for bd-110.","status":"open","priority":0,"issue_type":"task","created_at":"2025-10-16T22:47:49.785525-07:00","updated_at":"2025-10-16T22:47:49.785525-07:00","dependencies":[{"issue_id":"bd-114","depends_on_id":"bd-110","type":"parent-child","created_at":"2025-10-16T22:47:49.787472-07:00","created_by":"stevey"}]}
|
||||
{"id":"bd-12","title":"Root issue for dep tree test","description":"","status":"closed","priority":1,"issue_type":"task","created_at":"2025-10-16T20:46:08.971822-07:00","updated_at":"2025-10-16T21:51:08.743864-07:00","closed_at":"2025-10-16T10:07:34.1266-07:00"}
|
||||
{"id":"bd-13","title":"Dependency A","description":"","status":"closed","priority":1,"issue_type":"task","created_at":"2025-10-16T20:46:08.971822-07:00","updated_at":"2025-10-16T21:51:08.74444-07:00","closed_at":"2025-10-16T10:07:34.126732-07:00"}
|
||||
{"id":"bd-14","title":"Dependency B","description":"","status":"closed","priority":1,"issue_type":"task","created_at":"2025-10-16T20:46:08.971822-07:00","updated_at":"2025-10-16T21:51:08.745041-07:00","closed_at":"2025-10-16T10:07:34.126858-07:00"}
|
||||
|
||||
@@ -6,57 +6,58 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Client is an RPC client that communicates with the bd daemon.
|
||||
// Client represents an RPC client that connects to the daemon
|
||||
type Client struct {
|
||||
sockPath string
|
||||
mu sync.Mutex
|
||||
conn net.Conn
|
||||
conn net.Conn
|
||||
socketPath string
|
||||
}
|
||||
|
||||
// TryConnect attempts to connect to the daemon and returns a client if successful.
|
||||
// Returns nil if the daemon is not running or socket doesn't exist.
|
||||
func TryConnect(sockPath string) *Client {
|
||||
if _, err := os.Stat(sockPath); os.IsNotExist(err) {
|
||||
return nil
|
||||
// TryConnect attempts to connect to the daemon socket
|
||||
// Returns nil if no daemon is running
|
||||
func TryConnect(socketPath string) (*Client, error) {
|
||||
if _, err := os.Stat(socketPath); os.IsNotExist(err) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
conn, err := net.DialTimeout("unix", sockPath, 2*time.Second)
|
||||
conn, err := net.DialTimeout("unix", socketPath, 2*time.Second)
|
||||
if err != nil {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
client := &Client{
|
||||
sockPath: sockPath,
|
||||
conn: conn,
|
||||
conn: conn,
|
||||
socketPath: socketPath,
|
||||
}
|
||||
|
||||
if !client.ping() {
|
||||
if err := client.Ping(); err != nil {
|
||||
conn.Close()
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return client
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// ping sends a test request to verify the daemon is responsive.
|
||||
func (c *Client) ping() bool {
|
||||
req, _ := NewRequest(OpStats, nil)
|
||||
_, err := c.Execute(req)
|
||||
return err == nil
|
||||
// Close closes the connection to the daemon
|
||||
func (c *Client) Close() error {
|
||||
if c.conn != nil {
|
||||
return c.conn.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Execute sends a request to the daemon and returns the response.
|
||||
func (c *Client) Execute(req *Request) (*Response, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
// Execute sends an RPC request and waits for a response
|
||||
func (c *Client) Execute(operation string, args interface{}) (*Response, error) {
|
||||
argsJSON, err := json.Marshal(args)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal args: %w", err)
|
||||
}
|
||||
|
||||
if c.conn == nil {
|
||||
return nil, fmt.Errorf("client not connected")
|
||||
req := Request{
|
||||
Operation: operation,
|
||||
Args: argsJSON,
|
||||
}
|
||||
|
||||
reqJSON, err := json.Marshal(req)
|
||||
@@ -64,82 +65,100 @@ func (c *Client) Execute(req *Request) (*Response, error) {
|
||||
return nil, fmt.Errorf("failed to marshal request: %w", err)
|
||||
}
|
||||
|
||||
reqJSON = append(reqJSON, '\n')
|
||||
|
||||
if _, err := c.conn.Write(reqJSON); err != nil {
|
||||
c.reconnect()
|
||||
writer := bufio.NewWriter(c.conn)
|
||||
if _, err := writer.Write(reqJSON); err != nil {
|
||||
return nil, fmt.Errorf("failed to write request: %w", err)
|
||||
}
|
||||
if err := writer.WriteByte('\n'); err != nil {
|
||||
return nil, fmt.Errorf("failed to write newline: %w", err)
|
||||
}
|
||||
if err := writer.Flush(); err != nil {
|
||||
return nil, fmt.Errorf("failed to flush: %w", err)
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(c.conn)
|
||||
if !scanner.Scan() {
|
||||
if err := scanner.Err(); err != nil {
|
||||
c.reconnect()
|
||||
return nil, fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
c.reconnect()
|
||||
return nil, fmt.Errorf("connection closed")
|
||||
reader := bufio.NewReader(c.conn)
|
||||
respLine, err := reader.ReadBytes('\n')
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
|
||||
var resp Response
|
||||
if err := json.Unmarshal(scanner.Bytes(), &resp); err != nil {
|
||||
if err := json.Unmarshal(respLine, &resp); err != nil {
|
||||
return nil, fmt.Errorf("failed to unmarshal response: %w", err)
|
||||
}
|
||||
|
||||
if !resp.Success {
|
||||
return &resp, fmt.Errorf("operation failed: %s", resp.Error)
|
||||
}
|
||||
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
// reconnect attempts to reconnect to the daemon.
|
||||
func (c *Client) reconnect() error {
|
||||
if c.conn != nil {
|
||||
c.conn.Close()
|
||||
c.conn = nil
|
||||
}
|
||||
|
||||
var err error
|
||||
backoff := 100 * time.Millisecond
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
c.conn, err = net.DialTimeout("unix", c.sockPath, 2*time.Second)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
time.Sleep(backoff)
|
||||
backoff *= 2
|
||||
}
|
||||
|
||||
return fmt.Errorf("failed to reconnect after 3 attempts: %w", err)
|
||||
}
|
||||
|
||||
// Close closes the client connection.
|
||||
func (c *Client) Close() error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
if c.conn != nil {
|
||||
err := c.conn.Close()
|
||||
c.conn = nil
|
||||
// Ping sends a ping request to verify the daemon is alive
|
||||
func (c *Client) Ping() error {
|
||||
resp, err := c.Execute(OpPing, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if !resp.Success {
|
||||
return fmt.Errorf("ping failed: %s", resp.Error)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// SocketPath returns the default socket path for the given beads directory.
|
||||
func SocketPath(beadsDir string) string {
|
||||
return filepath.Join(beadsDir, "bd.sock")
|
||||
// Create creates a new issue via the daemon
|
||||
func (c *Client) Create(args *CreateArgs) (*Response, error) {
|
||||
return c.Execute(OpCreate, args)
|
||||
}
|
||||
|
||||
// DefaultSocketPath returns the socket path in the current working directory's .beads folder.
|
||||
func DefaultSocketPath() (string, error) {
|
||||
wd, err := os.Getwd()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to get working directory: %w", err)
|
||||
}
|
||||
|
||||
beadsDir := filepath.Join(wd, ".beads")
|
||||
if _, err := os.Stat(beadsDir); os.IsNotExist(err) {
|
||||
return "", fmt.Errorf(".beads directory not found")
|
||||
}
|
||||
|
||||
return SocketPath(beadsDir), nil
|
||||
// Update updates an issue via the daemon
|
||||
func (c *Client) Update(args *UpdateArgs) (*Response, error) {
|
||||
return c.Execute(OpUpdate, args)
|
||||
}
|
||||
|
||||
// Close closes an issue via the daemon (operation, not connection)
|
||||
func (c *Client) CloseIssue(args *CloseArgs) (*Response, error) {
|
||||
return c.Execute(OpClose, args)
|
||||
}
|
||||
|
||||
// List lists issues via the daemon
|
||||
func (c *Client) List(args *ListArgs) (*Response, error) {
|
||||
return c.Execute(OpList, args)
|
||||
}
|
||||
|
||||
// Show shows an issue via the daemon
|
||||
func (c *Client) Show(args *ShowArgs) (*Response, error) {
|
||||
return c.Execute(OpShow, args)
|
||||
}
|
||||
|
||||
// Ready gets ready work via the daemon
|
||||
func (c *Client) Ready(args *ReadyArgs) (*Response, error) {
|
||||
return c.Execute(OpReady, args)
|
||||
}
|
||||
|
||||
// Stats gets statistics via the daemon
|
||||
func (c *Client) Stats() (*Response, error) {
|
||||
return c.Execute(OpStats, nil)
|
||||
}
|
||||
|
||||
// AddDependency adds a dependency via the daemon
|
||||
func (c *Client) AddDependency(args *DepAddArgs) (*Response, error) {
|
||||
return c.Execute(OpDepAdd, args)
|
||||
}
|
||||
|
||||
// RemoveDependency removes a dependency via the daemon
|
||||
func (c *Client) RemoveDependency(args *DepRemoveArgs) (*Response, error) {
|
||||
return c.Execute(OpDepRemove, args)
|
||||
}
|
||||
|
||||
// AddLabel adds a label via the daemon
|
||||
func (c *Client) AddLabel(args *LabelAddArgs) (*Response, error) {
|
||||
return c.Execute(OpLabelAdd, args)
|
||||
}
|
||||
|
||||
// RemoveLabel removes a label via the daemon
|
||||
func (c *Client) RemoveLabel(args *LabelRemoveArgs) (*Response, error) {
|
||||
return c.Execute(OpLabelRemove, args)
|
||||
}
|
||||
|
||||
@@ -1,115 +0,0 @@
|
||||
package rpc
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestTryConnectNoSocket(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "nonexistent.sock")
|
||||
|
||||
client := TryConnect(sockPath)
|
||||
if client != nil {
|
||||
t.Error("Expected nil client when socket doesn't exist")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTryConnectSuccess(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "test.sock")
|
||||
|
||||
store := &mockStorage{}
|
||||
server := NewServer(store, sockPath)
|
||||
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Failed to start server: %v", err)
|
||||
}
|
||||
defer server.Stop()
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
client := TryConnect(sockPath)
|
||||
if client == nil {
|
||||
t.Fatal("Expected client to connect successfully")
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
if client.sockPath != sockPath {
|
||||
t.Errorf("Expected sockPath %s, got %s", sockPath, client.sockPath)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientExecute(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "test.sock")
|
||||
|
||||
store := &mockStorage{}
|
||||
server := NewServer(store, sockPath)
|
||||
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Failed to start server: %v", err)
|
||||
}
|
||||
defer server.Stop()
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
client := TryConnect(sockPath)
|
||||
if client == nil {
|
||||
t.Fatal("Failed to connect to server")
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
req, _ := NewRequest(OpList, map[string]string{"status": "open"})
|
||||
resp, err := client.Execute(req)
|
||||
if err != nil {
|
||||
t.Fatalf("Execute failed: %v", err)
|
||||
}
|
||||
|
||||
if resp.Success {
|
||||
t.Error("Expected error response for unimplemented operation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientMultipleRequests(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "test.sock")
|
||||
|
||||
store := &mockStorage{}
|
||||
server := NewServer(store, sockPath)
|
||||
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Failed to start server: %v", err)
|
||||
}
|
||||
defer server.Stop()
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
client := TryConnect(sockPath)
|
||||
if client == nil {
|
||||
t.Fatal("Failed to connect to server")
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
req, _ := NewRequest(OpStats, nil)
|
||||
resp, err := client.Execute(req)
|
||||
if err != nil {
|
||||
t.Fatalf("Execute %d failed: %v", i, err)
|
||||
}
|
||||
if resp == nil {
|
||||
t.Fatalf("Execute %d returned nil response", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSocketPath(t *testing.T) {
|
||||
beadsDir := "/home/user/project/.beads"
|
||||
expected := "/home/user/project/.beads/bd.sock"
|
||||
|
||||
result := SocketPath(beadsDir)
|
||||
if result != expected {
|
||||
t.Errorf("Expected %s, got %s", expected, result)
|
||||
}
|
||||
}
|
||||
@@ -2,87 +2,127 @@ package rpc
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
// Request represents an RPC request from a client to the daemon.
|
||||
// Operation constants for all bd commands
|
||||
const (
|
||||
OpPing = "ping"
|
||||
OpCreate = "create"
|
||||
OpUpdate = "update"
|
||||
OpClose = "close"
|
||||
OpList = "list"
|
||||
OpShow = "show"
|
||||
OpReady = "ready"
|
||||
OpStats = "stats"
|
||||
OpDepAdd = "dep_add"
|
||||
OpDepRemove = "dep_remove"
|
||||
OpDepTree = "dep_tree"
|
||||
OpLabelAdd = "label_add"
|
||||
OpLabelRemove = "label_remove"
|
||||
)
|
||||
|
||||
// Request represents an RPC request from client to daemon
|
||||
type Request struct {
|
||||
Operation string `json:"operation"`
|
||||
Args json.RawMessage `json:"args"`
|
||||
Actor string `json:"actor,omitempty"`
|
||||
RequestID string `json:"request_id,omitempty"`
|
||||
}
|
||||
|
||||
// Response represents an RPC response from the daemon to a client.
|
||||
// Response represents an RPC response from daemon to client
|
||||
type Response struct {
|
||||
Success bool `json:"success"`
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// BatchRequest represents a batch of operations to execute atomically.
|
||||
type BatchRequest struct {
|
||||
Operations []Request `json:"operations"`
|
||||
Atomic bool `json:"atomic"`
|
||||
// CreateArgs represents arguments for the create operation
|
||||
type CreateArgs struct {
|
||||
ID string `json:"id,omitempty"`
|
||||
Title string `json:"title"`
|
||||
Description string `json:"description,omitempty"`
|
||||
IssueType string `json:"issue_type"`
|
||||
Priority int `json:"priority"`
|
||||
Design string `json:"design,omitempty"`
|
||||
AcceptanceCriteria string `json:"acceptance_criteria,omitempty"`
|
||||
Assignee string `json:"assignee,omitempty"`
|
||||
Labels []string `json:"labels,omitempty"`
|
||||
Dependencies []string `json:"dependencies,omitempty"`
|
||||
}
|
||||
|
||||
// Operations supported by the daemon.
|
||||
const (
|
||||
OpCreate = "create"
|
||||
OpUpdate = "update"
|
||||
OpClose = "close"
|
||||
OpList = "list"
|
||||
OpShow = "show"
|
||||
OpReady = "ready"
|
||||
OpBlocked = "blocked"
|
||||
OpStats = "stats"
|
||||
OpDepAdd = "dep_add"
|
||||
OpDepRemove = "dep_remove"
|
||||
OpDepTree = "dep_tree"
|
||||
OpLabelAdd = "label_add"
|
||||
OpLabelRemove = "label_remove"
|
||||
OpLabelList = "label_list"
|
||||
OpLabelListAll = "label_list_all"
|
||||
OpExport = "export"
|
||||
OpImport = "import"
|
||||
OpCompact = "compact"
|
||||
OpRestore = "restore"
|
||||
OpBatch = "batch"
|
||||
)
|
||||
|
||||
// NewRequest creates a new RPC request with the given operation and arguments.
|
||||
func NewRequest(operation string, args interface{}) (*Request, error) {
|
||||
argsJSON, err := json.Marshal(args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Request{
|
||||
Operation: operation,
|
||||
Args: argsJSON,
|
||||
}, nil
|
||||
// UpdateArgs represents arguments for the update operation
|
||||
type UpdateArgs struct {
|
||||
ID string `json:"id"`
|
||||
Title *string `json:"title,omitempty"`
|
||||
Status *string `json:"status,omitempty"`
|
||||
Priority *int `json:"priority,omitempty"`
|
||||
Design *string `json:"design,omitempty"`
|
||||
AcceptanceCriteria *string `json:"acceptance_criteria,omitempty"`
|
||||
Notes *string `json:"notes,omitempty"`
|
||||
Assignee *string `json:"assignee,omitempty"`
|
||||
}
|
||||
|
||||
// NewSuccessResponse creates a successful response with the given data.
|
||||
func NewSuccessResponse(data interface{}) (*Response, error) {
|
||||
dataJSON, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Response{
|
||||
Success: true,
|
||||
Data: dataJSON,
|
||||
}, nil
|
||||
// CloseArgs represents arguments for the close operation
|
||||
type CloseArgs struct {
|
||||
ID string `json:"id"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
|
||||
// NewErrorResponse creates an error response with the given error message.
|
||||
func NewErrorResponse(err error) *Response {
|
||||
return &Response{
|
||||
Success: false,
|
||||
Error: err.Error(),
|
||||
}
|
||||
// ListArgs represents arguments for the list operation
|
||||
type ListArgs struct {
|
||||
Query string `json:"query,omitempty"`
|
||||
Status string `json:"status,omitempty"`
|
||||
Priority *int `json:"priority,omitempty"`
|
||||
IssueType string `json:"issue_type,omitempty"`
|
||||
Assignee string `json:"assignee,omitempty"`
|
||||
Label string `json:"label,omitempty"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
// UnmarshalArgs unmarshals the request arguments into the given value.
|
||||
func (r *Request) UnmarshalArgs(v interface{}) error {
|
||||
return json.Unmarshal(r.Args, v)
|
||||
// ShowArgs represents arguments for the show operation
|
||||
type ShowArgs struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
// UnmarshalData unmarshals the response data into the given value.
|
||||
func (r *Response) UnmarshalData(v interface{}) error {
|
||||
return json.Unmarshal(r.Data, v)
|
||||
// ReadyArgs represents arguments for the ready operation
|
||||
type ReadyArgs struct {
|
||||
Assignee string `json:"assignee,omitempty"`
|
||||
Priority *int `json:"priority,omitempty"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
}
|
||||
|
||||
// DepAddArgs represents arguments for adding a dependency
|
||||
type DepAddArgs struct {
|
||||
FromID string `json:"from_id"`
|
||||
ToID string `json:"to_id"`
|
||||
DepType string `json:"dep_type"`
|
||||
}
|
||||
|
||||
// DepRemoveArgs represents arguments for removing a dependency
|
||||
type DepRemoveArgs struct {
|
||||
FromID string `json:"from_id"`
|
||||
ToID string `json:"to_id"`
|
||||
DepType string `json:"dep_type,omitempty"`
|
||||
}
|
||||
|
||||
// DepTreeArgs represents arguments for the dep tree operation
|
||||
type DepTreeArgs struct {
|
||||
ID string `json:"id"`
|
||||
MaxDepth int `json:"max_depth,omitempty"`
|
||||
}
|
||||
|
||||
// LabelAddArgs represents arguments for adding a label
|
||||
type LabelAddArgs struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
}
|
||||
|
||||
// LabelRemoveArgs represents arguments for removing a label
|
||||
type LabelRemoveArgs struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
}
|
||||
|
||||
// PingResponse is the response for a ping operation
|
||||
type PingResponse struct {
|
||||
Message string `json:"message"`
|
||||
Version string `json:"version"`
|
||||
}
|
||||
|
||||
@@ -2,146 +2,169 @@ package rpc
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestNewRequest(t *testing.T) {
|
||||
args := map[string]string{
|
||||
"title": "Test issue",
|
||||
"priority": "1",
|
||||
func TestRequestSerialization(t *testing.T) {
|
||||
createArgs := CreateArgs{
|
||||
Title: "Test Issue",
|
||||
Description: "Test description",
|
||||
IssueType: "task",
|
||||
Priority: 2,
|
||||
}
|
||||
|
||||
req, err := NewRequest(OpCreate, args)
|
||||
argsJSON, err := json.Marshal(createArgs)
|
||||
if err != nil {
|
||||
t.Fatalf("NewRequest failed: %v", err)
|
||||
t.Fatalf("Failed to marshal args: %v", err)
|
||||
}
|
||||
|
||||
if req.Operation != OpCreate {
|
||||
t.Errorf("Expected operation %s, got %s", OpCreate, req.Operation)
|
||||
}
|
||||
|
||||
var unmarshaledArgs map[string]string
|
||||
if err := req.UnmarshalArgs(&unmarshaledArgs); err != nil {
|
||||
t.Fatalf("UnmarshalArgs failed: %v", err)
|
||||
}
|
||||
|
||||
if unmarshaledArgs["title"] != args["title"] {
|
||||
t.Errorf("Expected title %s, got %s", args["title"], unmarshaledArgs["title"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewSuccessResponse(t *testing.T) {
|
||||
data := map[string]interface{}{
|
||||
"id": "bd-123",
|
||||
"status": "open",
|
||||
}
|
||||
|
||||
resp, err := NewSuccessResponse(data)
|
||||
if err != nil {
|
||||
t.Fatalf("NewSuccessResponse failed: %v", err)
|
||||
}
|
||||
|
||||
if !resp.Success {
|
||||
t.Error("Expected success=true")
|
||||
}
|
||||
|
||||
if resp.Error != "" {
|
||||
t.Errorf("Expected empty error, got %s", resp.Error)
|
||||
}
|
||||
|
||||
var unmarshaledData map[string]interface{}
|
||||
if err := resp.UnmarshalData(&unmarshaledData); err != nil {
|
||||
t.Fatalf("UnmarshalData failed: %v", err)
|
||||
}
|
||||
|
||||
if unmarshaledData["id"] != data["id"] {
|
||||
t.Errorf("Expected id %s, got %s", data["id"], unmarshaledData["id"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewErrorResponse(t *testing.T) {
|
||||
testErr := errors.New("test error")
|
||||
|
||||
resp := NewErrorResponse(testErr)
|
||||
|
||||
if resp.Success {
|
||||
t.Error("Expected success=false")
|
||||
}
|
||||
|
||||
if resp.Error != testErr.Error() {
|
||||
t.Errorf("Expected error %s, got %s", testErr.Error(), resp.Error)
|
||||
}
|
||||
|
||||
if len(resp.Data) != 0 {
|
||||
t.Errorf("Expected empty data, got %v", resp.Data)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestResponseJSON(t *testing.T) {
|
||||
req := &Request{
|
||||
Operation: OpList,
|
||||
Args: json.RawMessage(`{"status":"open"}`),
|
||||
req := Request{
|
||||
Operation: OpCreate,
|
||||
Args: argsJSON,
|
||||
}
|
||||
|
||||
reqJSON, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal request failed: %v", err)
|
||||
t.Fatalf("Failed to marshal request: %v", err)
|
||||
}
|
||||
|
||||
var unmarshaledReq Request
|
||||
if err := json.Unmarshal(reqJSON, &unmarshaledReq); err != nil {
|
||||
t.Fatalf("Unmarshal request failed: %v", err)
|
||||
var decodedReq Request
|
||||
if err := json.Unmarshal(reqJSON, &decodedReq); err != nil {
|
||||
t.Fatalf("Failed to unmarshal request: %v", err)
|
||||
}
|
||||
|
||||
if unmarshaledReq.Operation != req.Operation {
|
||||
t.Errorf("Expected operation %s, got %s", req.Operation, unmarshaledReq.Operation)
|
||||
if decodedReq.Operation != OpCreate {
|
||||
t.Errorf("Expected operation %s, got %s", OpCreate, decodedReq.Operation)
|
||||
}
|
||||
|
||||
resp := &Response{
|
||||
var decodedArgs CreateArgs
|
||||
if err := json.Unmarshal(decodedReq.Args, &decodedArgs); err != nil {
|
||||
t.Fatalf("Failed to unmarshal args: %v", err)
|
||||
}
|
||||
|
||||
if decodedArgs.Title != createArgs.Title {
|
||||
t.Errorf("Expected title %s, got %s", createArgs.Title, decodedArgs.Title)
|
||||
}
|
||||
if decodedArgs.Priority != createArgs.Priority {
|
||||
t.Errorf("Expected priority %d, got %d", createArgs.Priority, decodedArgs.Priority)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseSerialization(t *testing.T) {
|
||||
resp := Response{
|
||||
Success: true,
|
||||
Data: json.RawMessage(`{"count":5}`),
|
||||
Data: json.RawMessage(`{"id":"bd-1","title":"Test"}`),
|
||||
}
|
||||
|
||||
respJSON, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal response failed: %v", err)
|
||||
t.Fatalf("Failed to marshal response: %v", err)
|
||||
}
|
||||
|
||||
var unmarshaledResp Response
|
||||
if err := json.Unmarshal(respJSON, &unmarshaledResp); err != nil {
|
||||
t.Fatalf("Unmarshal response failed: %v", err)
|
||||
var decodedResp Response
|
||||
if err := json.Unmarshal(respJSON, &decodedResp); err != nil {
|
||||
t.Fatalf("Failed to unmarshal response: %v", err)
|
||||
}
|
||||
|
||||
if unmarshaledResp.Success != resp.Success {
|
||||
t.Errorf("Expected success %v, got %v", resp.Success, unmarshaledResp.Success)
|
||||
if decodedResp.Success != resp.Success {
|
||||
t.Errorf("Expected success %v, got %v", resp.Success, decodedResp.Success)
|
||||
}
|
||||
|
||||
if string(decodedResp.Data) != string(resp.Data) {
|
||||
t.Errorf("Expected data %s, got %s", string(resp.Data), string(decodedResp.Data))
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchRequest(t *testing.T) {
|
||||
req1, _ := NewRequest(OpCreate, map[string]string{"title": "Issue 1"})
|
||||
req2, _ := NewRequest(OpCreate, map[string]string{"title": "Issue 2"})
|
||||
|
||||
batch := &BatchRequest{
|
||||
Operations: []Request{*req1, *req2},
|
||||
Atomic: true,
|
||||
func TestErrorResponse(t *testing.T) {
|
||||
resp := Response{
|
||||
Success: false,
|
||||
Error: "something went wrong",
|
||||
}
|
||||
|
||||
batchJSON, err := json.Marshal(batch)
|
||||
respJSON, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal batch failed: %v", err)
|
||||
t.Fatalf("Failed to marshal response: %v", err)
|
||||
}
|
||||
|
||||
var unmarshaledBatch BatchRequest
|
||||
if err := json.Unmarshal(batchJSON, &unmarshaledBatch); err != nil {
|
||||
t.Fatalf("Unmarshal batch failed: %v", err)
|
||||
var decodedResp Response
|
||||
if err := json.Unmarshal(respJSON, &decodedResp); err != nil {
|
||||
t.Fatalf("Failed to unmarshal response: %v", err)
|
||||
}
|
||||
|
||||
if len(unmarshaledBatch.Operations) != 2 {
|
||||
t.Errorf("Expected 2 operations, got %d", len(unmarshaledBatch.Operations))
|
||||
if decodedResp.Success {
|
||||
t.Errorf("Expected success false, got true")
|
||||
}
|
||||
|
||||
if !unmarshaledBatch.Atomic {
|
||||
t.Error("Expected atomic=true")
|
||||
if decodedResp.Error != resp.Error {
|
||||
t.Errorf("Expected error %s, got %s", resp.Error, decodedResp.Error)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllOperations(t *testing.T) {
|
||||
operations := []string{
|
||||
OpPing,
|
||||
OpCreate,
|
||||
OpUpdate,
|
||||
OpClose,
|
||||
OpList,
|
||||
OpShow,
|
||||
OpReady,
|
||||
OpStats,
|
||||
OpDepAdd,
|
||||
OpDepRemove,
|
||||
OpDepTree,
|
||||
OpLabelAdd,
|
||||
OpLabelRemove,
|
||||
}
|
||||
|
||||
for _, op := range operations {
|
||||
req := Request{
|
||||
Operation: op,
|
||||
Args: json.RawMessage(`{}`),
|
||||
}
|
||||
|
||||
reqJSON, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
t.Errorf("Failed to marshal request for op %s: %v", op, err)
|
||||
continue
|
||||
}
|
||||
|
||||
var decodedReq Request
|
||||
if err := json.Unmarshal(reqJSON, &decodedReq); err != nil {
|
||||
t.Errorf("Failed to unmarshal request for op %s: %v", op, err)
|
||||
continue
|
||||
}
|
||||
|
||||
if decodedReq.Operation != op {
|
||||
t.Errorf("Expected operation %s, got %s", op, decodedReq.Operation)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateArgsWithNilValues(t *testing.T) {
|
||||
title := "New Title"
|
||||
args := UpdateArgs{
|
||||
ID: "bd-1",
|
||||
Title: &title,
|
||||
}
|
||||
|
||||
argsJSON, err := json.Marshal(args)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to marshal args: %v", err)
|
||||
}
|
||||
|
||||
var decodedArgs UpdateArgs
|
||||
if err := json.Unmarshal(argsJSON, &decodedArgs); err != nil {
|
||||
t.Fatalf("Failed to unmarshal args: %v", err)
|
||||
}
|
||||
|
||||
if decodedArgs.Title == nil {
|
||||
t.Errorf("Expected title to be non-nil")
|
||||
} else if *decodedArgs.Title != title {
|
||||
t.Errorf("Expected title %s, got %s", title, *decodedArgs.Title)
|
||||
}
|
||||
|
||||
if decodedArgs.Status != nil {
|
||||
t.Errorf("Expected status to be nil, got %v", *decodedArgs.Status)
|
||||
}
|
||||
}
|
||||
|
||||
243
internal/rpc/rpc_test.go
Normal file
243
internal/rpc/rpc_test.go
Normal file
@@ -0,0 +1,243 @@
|
||||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
sqlitestorage "github.com/steveyegge/beads/internal/storage/sqlite"
|
||||
"github.com/steveyegge/beads/internal/types"
|
||||
)
|
||||
|
||||
func setupTestServer(t *testing.T) (*Server, *Client, func()) {
|
||||
tmpDir, err := os.MkdirTemp("", "bd-rpc-test-*")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create temp dir: %v", err)
|
||||
}
|
||||
|
||||
dbPath := filepath.Join(tmpDir, "test.db")
|
||||
socketPath := filepath.Join(tmpDir, "bd.sock")
|
||||
|
||||
store, err := sqlitestorage.New(dbPath)
|
||||
if err != nil {
|
||||
os.RemoveAll(tmpDir)
|
||||
t.Fatalf("Failed to create store: %v", err)
|
||||
}
|
||||
|
||||
server := NewServer(socketPath, store)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
if err := server.Start(ctx); err != nil && err.Error() != "accept unix "+socketPath+": use of closed network connection" {
|
||||
t.Logf("Server error: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
client, err := TryConnect(socketPath)
|
||||
if err != nil {
|
||||
cancel()
|
||||
server.Stop()
|
||||
store.Close()
|
||||
os.RemoveAll(tmpDir)
|
||||
t.Fatalf("Failed to connect client: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
client.Close()
|
||||
cancel()
|
||||
server.Stop()
|
||||
store.Close()
|
||||
os.RemoveAll(tmpDir)
|
||||
}
|
||||
|
||||
return server, client, cleanup
|
||||
}
|
||||
|
||||
func TestPing(t *testing.T) {
|
||||
_, client, cleanup := setupTestServer(t)
|
||||
defer cleanup()
|
||||
|
||||
if err := client.Ping(); err != nil {
|
||||
t.Fatalf("Ping failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateIssue(t *testing.T) {
|
||||
_, client, cleanup := setupTestServer(t)
|
||||
defer cleanup()
|
||||
|
||||
args := &CreateArgs{
|
||||
Title: "Test Issue",
|
||||
Description: "Test description",
|
||||
IssueType: "task",
|
||||
Priority: 2,
|
||||
}
|
||||
|
||||
resp, err := client.Create(args)
|
||||
if err != nil {
|
||||
t.Fatalf("Create failed: %v", err)
|
||||
}
|
||||
|
||||
if !resp.Success {
|
||||
t.Fatalf("Expected success, got error: %s", resp.Error)
|
||||
}
|
||||
|
||||
var issue types.Issue
|
||||
if err := json.Unmarshal(resp.Data, &issue); err != nil {
|
||||
t.Fatalf("Failed to unmarshal issue: %v", err)
|
||||
}
|
||||
|
||||
if issue.Title != args.Title {
|
||||
t.Errorf("Expected title %s, got %s", args.Title, issue.Title)
|
||||
}
|
||||
if issue.Priority != args.Priority {
|
||||
t.Errorf("Expected priority %d, got %d", args.Priority, issue.Priority)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateIssue(t *testing.T) {
|
||||
_, client, cleanup := setupTestServer(t)
|
||||
defer cleanup()
|
||||
|
||||
createArgs := &CreateArgs{
|
||||
Title: "Original Title",
|
||||
IssueType: "task",
|
||||
Priority: 2,
|
||||
}
|
||||
|
||||
createResp, err := client.Create(createArgs)
|
||||
if err != nil {
|
||||
t.Fatalf("Create failed: %v", err)
|
||||
}
|
||||
|
||||
var issue types.Issue
|
||||
json.Unmarshal(createResp.Data, &issue)
|
||||
|
||||
newTitle := "Updated Title"
|
||||
updateArgs := &UpdateArgs{
|
||||
ID: issue.ID,
|
||||
Title: &newTitle,
|
||||
}
|
||||
|
||||
updateResp, err := client.Update(updateArgs)
|
||||
if err != nil {
|
||||
t.Fatalf("Update failed: %v", err)
|
||||
}
|
||||
|
||||
var updatedIssue types.Issue
|
||||
json.Unmarshal(updateResp.Data, &updatedIssue)
|
||||
|
||||
if updatedIssue.Title != newTitle {
|
||||
t.Errorf("Expected title %s, got %s", newTitle, updatedIssue.Title)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListIssues(t *testing.T) {
|
||||
_, client, cleanup := setupTestServer(t)
|
||||
defer cleanup()
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
args := &CreateArgs{
|
||||
Title: "Test Issue",
|
||||
IssueType: "task",
|
||||
Priority: 2,
|
||||
}
|
||||
if _, err := client.Create(args); err != nil {
|
||||
t.Fatalf("Create failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
listArgs := &ListArgs{
|
||||
Limit: 10,
|
||||
}
|
||||
|
||||
resp, err := client.List(listArgs)
|
||||
if err != nil {
|
||||
t.Fatalf("List failed: %v", err)
|
||||
}
|
||||
|
||||
var issues []types.Issue
|
||||
if err := json.Unmarshal(resp.Data, &issues); err != nil {
|
||||
t.Fatalf("Failed to unmarshal issues: %v", err)
|
||||
}
|
||||
|
||||
if len(issues) != 3 {
|
||||
t.Errorf("Expected 3 issues, got %d", len(issues))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSocketCleanup(t *testing.T) {
|
||||
tmpDir, err := os.MkdirTemp("", "bd-rpc-cleanup-test-*")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create temp dir: %v", err)
|
||||
}
|
||||
defer os.RemoveAll(tmpDir)
|
||||
|
||||
dbPath := filepath.Join(tmpDir, "test.db")
|
||||
socketPath := filepath.Join(tmpDir, "bd.sock")
|
||||
|
||||
store, err := sqlitestorage.New(dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create store: %v", err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
server := NewServer(socketPath, store)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go server.Start(ctx)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
if _, err := os.Stat(socketPath); os.IsNotExist(err) {
|
||||
t.Fatal("Socket file not created")
|
||||
}
|
||||
|
||||
if err := server.Stop(); err != nil {
|
||||
t.Fatalf("Stop failed: %v", err)
|
||||
}
|
||||
|
||||
if _, err := os.Stat(socketPath); !os.IsNotExist(err) {
|
||||
t.Fatal("Socket file not cleaned up")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConcurrentRequests(t *testing.T) {
|
||||
_, client, cleanup := setupTestServer(t)
|
||||
defer cleanup()
|
||||
|
||||
done := make(chan bool)
|
||||
errors := make(chan error, 5)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
go func(n int) {
|
||||
args := &CreateArgs{
|
||||
Title: "Concurrent Issue",
|
||||
IssueType: "task",
|
||||
Priority: 2,
|
||||
}
|
||||
|
||||
if _, err := client.Create(args); err != nil {
|
||||
errors <- err
|
||||
}
|
||||
done <- true
|
||||
}(i)
|
||||
}
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
<-done
|
||||
}
|
||||
|
||||
close(errors)
|
||||
for err := range errors {
|
||||
if err != nil {
|
||||
t.Errorf("Concurrent request failed: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -7,259 +7,552 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"syscall"
|
||||
|
||||
"github.com/steveyegge/beads/internal/storage"
|
||||
"github.com/steveyegge/beads/internal/types"
|
||||
)
|
||||
|
||||
// Server is the RPC server that handles requests from bd clients.
|
||||
// Server represents the RPC server that runs in the daemon
|
||||
type Server struct {
|
||||
storage storage.Storage
|
||||
listener net.Listener
|
||||
sockPath string
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex // Protects shutdown state
|
||||
shutdown bool
|
||||
socketPath string
|
||||
storage storage.Storage
|
||||
listener net.Listener
|
||||
mu sync.Mutex
|
||||
shutdown bool
|
||||
}
|
||||
|
||||
// NewServer creates a new RPC server.
|
||||
func NewServer(store storage.Storage, sockPath string) *Server {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
// NewServer creates a new RPC server
|
||||
func NewServer(socketPath string, store storage.Storage) *Server {
|
||||
return &Server{
|
||||
storage: store,
|
||||
sockPath: sockPath,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
socketPath: socketPath,
|
||||
storage: store,
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the RPC server listening on the Unix socket.
|
||||
func (s *Server) Start() error {
|
||||
if err := os.Remove(s.sockPath); err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("failed to remove existing socket: %w", err)
|
||||
// Start starts the RPC server and listens for connections
|
||||
func (s *Server) Start(ctx context.Context) error {
|
||||
if err := s.ensureSocketDir(); err != nil {
|
||||
return fmt.Errorf("failed to ensure socket directory: %w", err)
|
||||
}
|
||||
|
||||
listener, err := net.Listen("unix", s.sockPath)
|
||||
if err := s.removeOldSocket(); err != nil {
|
||||
return fmt.Errorf("failed to remove old socket: %w", err)
|
||||
}
|
||||
|
||||
var err error
|
||||
s.listener, err = net.Listen("unix", s.socketPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to listen on socket %s: %w", s.sockPath, err)
|
||||
}
|
||||
s.listener = listener
|
||||
|
||||
if err := os.Chmod(s.sockPath, 0600); err != nil {
|
||||
s.listener.Close()
|
||||
return fmt.Errorf("failed to set socket permissions: %w", err)
|
||||
return fmt.Errorf("failed to listen on socket: %w", err)
|
||||
}
|
||||
|
||||
s.wg.Add(1)
|
||||
go s.acceptLoop()
|
||||
go s.handleSignals()
|
||||
|
||||
return nil
|
||||
for {
|
||||
conn, err := s.listener.Accept()
|
||||
if err != nil {
|
||||
s.mu.Lock()
|
||||
shutdown := s.shutdown
|
||||
s.mu.Unlock()
|
||||
if shutdown {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("failed to accept connection: %w", err)
|
||||
}
|
||||
|
||||
go s.handleConnection(conn)
|
||||
}
|
||||
}
|
||||
|
||||
// Stop gracefully stops the RPC server.
|
||||
// Stop stops the RPC server and cleans up resources
|
||||
func (s *Server) Stop() error {
|
||||
s.mu.Lock()
|
||||
if s.shutdown {
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
s.shutdown = true
|
||||
s.mu.Unlock()
|
||||
|
||||
s.cancel()
|
||||
|
||||
if s.listener != nil {
|
||||
s.listener.Close()
|
||||
if err := s.listener.Close(); err != nil {
|
||||
return fmt.Errorf("failed to close listener: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
s.wg.Wait()
|
||||
|
||||
if err := os.Remove(s.sockPath); err != nil && !os.IsNotExist(err) {
|
||||
if err := s.removeOldSocket(); err != nil {
|
||||
return fmt.Errorf("failed to remove socket: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// acceptLoop accepts incoming connections and handles them.
|
||||
func (s *Server) acceptLoop() {
|
||||
defer s.wg.Done()
|
||||
|
||||
for {
|
||||
conn, err := s.listener.Accept()
|
||||
if err != nil {
|
||||
select {
|
||||
case <-s.ctx.Done():
|
||||
return
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "Error accepting connection: %v\n", err)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
s.wg.Add(1)
|
||||
go s.handleConnection(conn)
|
||||
func (s *Server) ensureSocketDir() error {
|
||||
dir := filepath.Dir(s.socketPath)
|
||||
if err := os.MkdirAll(dir, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) removeOldSocket() error {
|
||||
if _, err := os.Stat(s.socketPath); err == nil {
|
||||
if err := os.Remove(s.socketPath); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) handleSignals() {
|
||||
sigChan := make(chan os.Signal, 1)
|
||||
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-sigChan
|
||||
s.Stop()
|
||||
}
|
||||
|
||||
// handleConnection handles a single client connection.
|
||||
func (s *Server) handleConnection(conn net.Conn) {
|
||||
defer s.wg.Done()
|
||||
defer conn.Close()
|
||||
|
||||
scanner := bufio.NewScanner(conn)
|
||||
reader := bufio.NewReader(conn)
|
||||
writer := bufio.NewWriter(conn)
|
||||
|
||||
for scanner.Scan() {
|
||||
select {
|
||||
case <-s.ctx.Done():
|
||||
for {
|
||||
line, err := reader.ReadBytes('\n')
|
||||
if err != nil {
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
line := scanner.Bytes()
|
||||
var req Request
|
||||
if err := json.Unmarshal(line, &req); err != nil {
|
||||
resp := NewErrorResponse(fmt.Errorf("invalid request JSON: %w", err))
|
||||
s.sendResponse(writer, resp)
|
||||
resp := Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid request: %v", err),
|
||||
}
|
||||
s.writeResponse(writer, resp)
|
||||
continue
|
||||
}
|
||||
|
||||
resp := s.handleRequest(&req)
|
||||
s.sendResponse(writer, resp)
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error reading from connection: %v\n", err)
|
||||
s.writeResponse(writer, resp)
|
||||
}
|
||||
}
|
||||
|
||||
// sendResponse sends a response to the client.
|
||||
func (s *Server) sendResponse(writer *bufio.Writer, resp *Response) {
|
||||
respJSON, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error marshaling response: %v\n", err)
|
||||
return
|
||||
}
|
||||
|
||||
if _, err := writer.Write(respJSON); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error writing response: %v\n", err)
|
||||
return
|
||||
}
|
||||
if _, err := writer.Write([]byte("\n")); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error writing newline: %v\n", err)
|
||||
return
|
||||
}
|
||||
if err := writer.Flush(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Error flushing response: %v\n", err)
|
||||
}
|
||||
}
|
||||
|
||||
// handleRequest processes an RPC request and returns a response.
|
||||
func (s *Server) handleRequest(req *Request) *Response {
|
||||
ctx := context.Background()
|
||||
|
||||
func (s *Server) handleRequest(req *Request) Response {
|
||||
switch req.Operation {
|
||||
case OpBatch:
|
||||
return s.handleBatch(ctx, req)
|
||||
case OpPing:
|
||||
return s.handlePing(req)
|
||||
case OpCreate:
|
||||
return s.handleCreate(ctx, req)
|
||||
return s.handleCreate(req)
|
||||
case OpUpdate:
|
||||
return s.handleUpdate(ctx, req)
|
||||
return s.handleUpdate(req)
|
||||
case OpClose:
|
||||
return s.handleClose(ctx, req)
|
||||
return s.handleClose(req)
|
||||
case OpList:
|
||||
return s.handleList(ctx, req)
|
||||
return s.handleList(req)
|
||||
case OpShow:
|
||||
return s.handleShow(ctx, req)
|
||||
return s.handleShow(req)
|
||||
case OpReady:
|
||||
return s.handleReady(ctx, req)
|
||||
case OpBlocked:
|
||||
return s.handleBlocked(ctx, req)
|
||||
return s.handleReady(req)
|
||||
case OpStats:
|
||||
return s.handleStats(ctx, req)
|
||||
return s.handleStats(req)
|
||||
case OpDepAdd:
|
||||
return s.handleDepAdd(ctx, req)
|
||||
return s.handleDepAdd(req)
|
||||
case OpDepRemove:
|
||||
return s.handleDepRemove(ctx, req)
|
||||
case OpDepTree:
|
||||
return s.handleDepTree(ctx, req)
|
||||
return s.handleDepRemove(req)
|
||||
case OpLabelAdd:
|
||||
return s.handleLabelAdd(ctx, req)
|
||||
return s.handleLabelAdd(req)
|
||||
case OpLabelRemove:
|
||||
return s.handleLabelRemove(ctx, req)
|
||||
case OpLabelList:
|
||||
return s.handleLabelList(ctx, req)
|
||||
case OpLabelListAll:
|
||||
return s.handleLabelListAll(ctx, req)
|
||||
return s.handleLabelRemove(req)
|
||||
default:
|
||||
return NewErrorResponse(fmt.Errorf("unknown operation: %s", req.Operation))
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("unknown operation: %s", req.Operation),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Placeholder handlers - will be implemented in future commits
|
||||
func (s *Server) handleBatch(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("batch operation not yet implemented"))
|
||||
// Adapter helpers
|
||||
func (s *Server) reqCtx(_ *Request) context.Context {
|
||||
return context.Background()
|
||||
}
|
||||
|
||||
func (s *Server) handleCreate(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("create operation not yet implemented"))
|
||||
func (s *Server) reqActor(req *Request) string {
|
||||
if req != nil && req.Actor != "" {
|
||||
return req.Actor
|
||||
}
|
||||
return "daemon"
|
||||
}
|
||||
|
||||
func (s *Server) handleUpdate(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("update operation not yet implemented"))
|
||||
func strValue(p *string) string {
|
||||
if p == nil {
|
||||
return ""
|
||||
}
|
||||
return *p
|
||||
}
|
||||
|
||||
func (s *Server) handleClose(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("close operation not yet implemented"))
|
||||
func strPtr(s string) *string {
|
||||
if s == "" {
|
||||
return nil
|
||||
}
|
||||
return &s
|
||||
}
|
||||
|
||||
func (s *Server) handleList(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("list operation not yet implemented"))
|
||||
func updatesFromArgs(a UpdateArgs) map[string]interface{} {
|
||||
u := map[string]interface{}{}
|
||||
if a.Title != nil {
|
||||
u["title"] = *a.Title
|
||||
}
|
||||
if a.Status != nil {
|
||||
u["status"] = *a.Status
|
||||
}
|
||||
if a.Priority != nil {
|
||||
u["priority"] = *a.Priority
|
||||
}
|
||||
if a.Design != nil {
|
||||
u["design"] = a.Design
|
||||
}
|
||||
if a.AcceptanceCriteria != nil {
|
||||
u["acceptance_criteria"] = a.AcceptanceCriteria
|
||||
}
|
||||
if a.Notes != nil {
|
||||
u["notes"] = a.Notes
|
||||
}
|
||||
if a.Assignee != nil {
|
||||
u["assignee"] = a.Assignee
|
||||
}
|
||||
return u
|
||||
}
|
||||
|
||||
func (s *Server) handleShow(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("show operation not yet implemented"))
|
||||
// Handler implementations
|
||||
|
||||
func (s *Server) handlePing(_ *Request) Response {
|
||||
data, _ := json.Marshal(PingResponse{
|
||||
Message: "pong",
|
||||
Version: "0.9.8",
|
||||
})
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleReady(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("ready operation not yet implemented"))
|
||||
func (s *Server) handleCreate(req *Request) Response {
|
||||
var createArgs CreateArgs
|
||||
if err := json.Unmarshal(req.Args, &createArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid create args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
var design, acceptance, assignee *string
|
||||
if createArgs.Design != "" {
|
||||
design = &createArgs.Design
|
||||
}
|
||||
if createArgs.AcceptanceCriteria != "" {
|
||||
acceptance = &createArgs.AcceptanceCriteria
|
||||
}
|
||||
if createArgs.Assignee != "" {
|
||||
assignee = &createArgs.Assignee
|
||||
}
|
||||
|
||||
issue := &types.Issue{
|
||||
ID: createArgs.ID,
|
||||
Title: createArgs.Title,
|
||||
Description: createArgs.Description,
|
||||
IssueType: types.IssueType(createArgs.IssueType),
|
||||
Priority: createArgs.Priority,
|
||||
Design: strValue(design),
|
||||
AcceptanceCriteria: strValue(acceptance),
|
||||
Assignee: strValue(assignee),
|
||||
Status: types.StatusOpen,
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
if err := s.storage.CreateIssue(ctx, issue, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to create issue: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(issue)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleBlocked(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("blocked operation not yet implemented"))
|
||||
func (s *Server) handleUpdate(req *Request) Response {
|
||||
var updateArgs UpdateArgs
|
||||
if err := json.Unmarshal(req.Args, &updateArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid update args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
updates := updatesFromArgs(updateArgs)
|
||||
if len(updates) == 0 {
|
||||
return Response{Success: true}
|
||||
}
|
||||
|
||||
if err := s.storage.UpdateIssue(ctx, updateArgs.ID, updates, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to update issue: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
issue, err := s.storage.GetIssue(ctx, updateArgs.ID)
|
||||
if err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to get updated issue: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(issue)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleStats(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("stats operation not yet implemented"))
|
||||
func (s *Server) handleClose(req *Request) Response {
|
||||
var closeArgs CloseArgs
|
||||
if err := json.Unmarshal(req.Args, &closeArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid close args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
if err := s.storage.CloseIssue(ctx, closeArgs.ID, closeArgs.Reason, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to close issue: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
issue, _ := s.storage.GetIssue(ctx, closeArgs.ID)
|
||||
data, _ := json.Marshal(issue)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleDepAdd(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("dep_add operation not yet implemented"))
|
||||
func (s *Server) handleList(req *Request) Response {
|
||||
var listArgs ListArgs
|
||||
if err := json.Unmarshal(req.Args, &listArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid list args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
filter := types.IssueFilter{
|
||||
Limit: listArgs.Limit,
|
||||
}
|
||||
if listArgs.Status != "" {
|
||||
status := types.Status(listArgs.Status)
|
||||
filter.Status = &status
|
||||
}
|
||||
if listArgs.IssueType != "" {
|
||||
issueType := types.IssueType(listArgs.IssueType)
|
||||
filter.IssueType = &issueType
|
||||
}
|
||||
if listArgs.Assignee != "" {
|
||||
filter.Assignee = &listArgs.Assignee
|
||||
}
|
||||
if listArgs.Priority != nil {
|
||||
filter.Priority = listArgs.Priority
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
issues, err := s.storage.SearchIssues(ctx, listArgs.Query, filter)
|
||||
if err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to list issues: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(issues)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleDepRemove(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("dep_remove operation not yet implemented"))
|
||||
func (s *Server) handleShow(req *Request) Response {
|
||||
var showArgs ShowArgs
|
||||
if err := json.Unmarshal(req.Args, &showArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid show args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
issue, err := s.storage.GetIssue(ctx, showArgs.ID)
|
||||
if err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to get issue: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(issue)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleDepTree(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("dep_tree operation not yet implemented"))
|
||||
func (s *Server) handleReady(req *Request) Response {
|
||||
var readyArgs ReadyArgs
|
||||
if err := json.Unmarshal(req.Args, &readyArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid ready args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
wf := types.WorkFilter{
|
||||
Status: types.StatusOpen,
|
||||
Priority: readyArgs.Priority,
|
||||
Limit: readyArgs.Limit,
|
||||
}
|
||||
if readyArgs.Assignee != "" {
|
||||
wf.Assignee = &readyArgs.Assignee
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
issues, err := s.storage.GetReadyWork(ctx, wf)
|
||||
if err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to get ready work: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(issues)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleLabelAdd(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("label_add operation not yet implemented"))
|
||||
func (s *Server) handleStats(req *Request) Response {
|
||||
ctx := s.reqCtx(req)
|
||||
stats, err := s.storage.GetStatistics(ctx)
|
||||
if err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to get statistics: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(stats)
|
||||
return Response{
|
||||
Success: true,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleLabelRemove(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("label_remove operation not yet implemented"))
|
||||
func (s *Server) handleDepAdd(req *Request) Response {
|
||||
var depArgs DepAddArgs
|
||||
if err := json.Unmarshal(req.Args, &depArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid dep add args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
dep := &types.Dependency{
|
||||
IssueID: depArgs.FromID,
|
||||
DependsOnID: depArgs.ToID,
|
||||
Type: types.DependencyType(depArgs.DepType),
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
if err := s.storage.AddDependency(ctx, dep, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to add dependency: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
return Response{Success: true}
|
||||
}
|
||||
|
||||
func (s *Server) handleLabelList(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("label_list operation not yet implemented"))
|
||||
func (s *Server) handleDepRemove(req *Request) Response {
|
||||
var depArgs DepRemoveArgs
|
||||
if err := json.Unmarshal(req.Args, &depArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid dep remove args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
if err := s.storage.RemoveDependency(ctx, depArgs.FromID, depArgs.ToID, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to remove dependency: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
return Response{Success: true}
|
||||
}
|
||||
|
||||
func (s *Server) handleLabelListAll(ctx context.Context, req *Request) *Response {
|
||||
return NewErrorResponse(fmt.Errorf("label_list_all operation not yet implemented"))
|
||||
func (s *Server) handleLabelAdd(req *Request) Response {
|
||||
var labelArgs LabelAddArgs
|
||||
if err := json.Unmarshal(req.Args, &labelArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid label add args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
if err := s.storage.AddLabel(ctx, labelArgs.ID, labelArgs.Label, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to add label: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
return Response{Success: true}
|
||||
}
|
||||
|
||||
func (s *Server) handleLabelRemove(req *Request) Response {
|
||||
var labelArgs LabelRemoveArgs
|
||||
if err := json.Unmarshal(req.Args, &labelArgs); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("invalid label remove args: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
ctx := s.reqCtx(req)
|
||||
if err := s.storage.RemoveLabel(ctx, labelArgs.ID, labelArgs.Label, s.reqActor(req)); err != nil {
|
||||
return Response{
|
||||
Success: false,
|
||||
Error: fmt.Sprintf("failed to remove label: %v", err),
|
||||
}
|
||||
}
|
||||
|
||||
return Response{Success: true}
|
||||
}
|
||||
|
||||
func (s *Server) writeResponse(writer *bufio.Writer, resp Response) {
|
||||
data, _ := json.Marshal(resp)
|
||||
writer.Write(data)
|
||||
writer.WriteByte('\n')
|
||||
writer.Flush()
|
||||
}
|
||||
|
||||
@@ -1,214 +0,0 @@
|
||||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/steveyegge/beads/internal/types"
|
||||
)
|
||||
|
||||
type mockStorage struct{}
|
||||
|
||||
func (m *mockStorage) CreateIssue(ctx context.Context, issue *types.Issue, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) CreateIssues(ctx context.Context, issues []*types.Issue, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) GetIssue(ctx context.Context, id string) (*types.Issue, error) { return nil, nil }
|
||||
func (m *mockStorage) UpdateIssue(ctx context.Context, id string, updates map[string]interface{}, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) CloseIssue(ctx context.Context, id, reason, actor string) error { return nil }
|
||||
func (m *mockStorage) SearchIssues(ctx context.Context, query string, filter types.IssueFilter) ([]*types.Issue, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) AddDependency(ctx context.Context, dep *types.Dependency, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) RemoveDependency(ctx context.Context, issueID, dependsOnID, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) GetDependencies(ctx context.Context, issueID string) ([]*types.Issue, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetDependents(ctx context.Context, issueID string) ([]*types.Issue, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetDependencyRecords(ctx context.Context, issueID string) ([]*types.Dependency, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetAllDependencyRecords(ctx context.Context) (map[string][]*types.Dependency, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetDependencyTree(ctx context.Context, issueID string, maxDepth int) ([]*types.TreeNode, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) DetectCycles(ctx context.Context) ([][]*types.Issue, error) { return nil, nil }
|
||||
func (m *mockStorage) AddLabel(ctx context.Context, issueID, label, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) RemoveLabel(ctx context.Context, issueID, label, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) GetLabels(ctx context.Context, issueID string) ([]string, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetIssuesByLabel(ctx context.Context, label string) ([]*types.Issue, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetReadyWork(ctx context.Context, filter types.WorkFilter) ([]*types.Issue, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetBlockedIssues(ctx context.Context) ([]*types.BlockedIssue, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) AddComment(ctx context.Context, issueID, actor, comment string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) GetEvents(ctx context.Context, issueID string, limit int) ([]*types.Event, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetStatistics(ctx context.Context) (*types.Statistics, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockStorage) GetDirtyIssues(ctx context.Context) ([]string, error) { return nil, nil }
|
||||
func (m *mockStorage) ClearDirtyIssues(ctx context.Context) error { return nil }
|
||||
func (m *mockStorage) ClearDirtyIssuesByID(ctx context.Context, issueIDs []string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) SetConfig(ctx context.Context, key, value string) error { return nil }
|
||||
func (m *mockStorage) GetConfig(ctx context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
func (m *mockStorage) SetMetadata(ctx context.Context, key, value string) error { return nil }
|
||||
func (m *mockStorage) GetMetadata(ctx context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
func (m *mockStorage) UpdateIssueID(ctx context.Context, oldID, newID string, issue *types.Issue, actor string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) RenameDependencyPrefix(ctx context.Context, oldPrefix, newPrefix string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) RenameCounterPrefix(ctx context.Context, oldPrefix, newPrefix string) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockStorage) Close() error { return nil }
|
||||
|
||||
func TestServerStartStop(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "test.sock")
|
||||
|
||||
store := &mockStorage{}
|
||||
server := NewServer(store, sockPath)
|
||||
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Failed to start server: %v", err)
|
||||
}
|
||||
|
||||
if _, err := os.Stat(sockPath); os.IsNotExist(err) {
|
||||
t.Fatal("Socket file was not created")
|
||||
}
|
||||
|
||||
if err := server.Stop(); err != nil {
|
||||
t.Fatalf("Failed to stop server: %v", err)
|
||||
}
|
||||
|
||||
if _, err := os.Stat(sockPath); !os.IsNotExist(err) {
|
||||
t.Fatal("Socket file was not removed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerHandlesRequest(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "test.sock")
|
||||
|
||||
store := &mockStorage{}
|
||||
server := NewServer(store, sockPath)
|
||||
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Failed to start server: %v", err)
|
||||
}
|
||||
defer server.Stop()
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
conn, err := net.Dial("unix", sockPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to connect to server: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
req, _ := NewRequest(OpStats, nil)
|
||||
reqJSON, _ := json.Marshal(req)
|
||||
reqJSON = append(reqJSON, '\n')
|
||||
|
||||
if _, err := conn.Write(reqJSON); err != nil {
|
||||
t.Fatalf("Failed to write request: %v", err)
|
||||
}
|
||||
|
||||
buf := make([]byte, 4096)
|
||||
n, err := conn.Read(buf)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to read response: %v", err)
|
||||
}
|
||||
|
||||
var resp Response
|
||||
if err := json.Unmarshal(buf[:n], &resp); err != nil {
|
||||
t.Fatalf("Failed to unmarshal response: %v", err)
|
||||
}
|
||||
|
||||
if resp.Success {
|
||||
t.Error("Expected error response for unimplemented operation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerRejectsInvalidJSON(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
sockPath := filepath.Join(tmpDir, "test.sock")
|
||||
|
||||
store := &mockStorage{}
|
||||
server := NewServer(store, sockPath)
|
||||
|
||||
if err := server.Start(); err != nil {
|
||||
t.Fatalf("Failed to start server: %v", err)
|
||||
}
|
||||
defer server.Stop()
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
conn, err := net.Dial("unix", sockPath)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to connect to server: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
if _, err := conn.Write([]byte("invalid json\n")); err != nil {
|
||||
t.Fatalf("Failed to write request: %v", err)
|
||||
}
|
||||
|
||||
buf := make([]byte, 4096)
|
||||
n, err := conn.Read(buf)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to read response: %v", err)
|
||||
}
|
||||
|
||||
var resp Response
|
||||
if err := json.Unmarshal(buf[:n], &resp); err != nil {
|
||||
t.Fatalf("Failed to unmarshal response: %v", err)
|
||||
}
|
||||
|
||||
if resp.Success {
|
||||
t.Error("Expected error response for invalid JSON")
|
||||
}
|
||||
|
||||
if resp.Error == "" {
|
||||
t.Error("Expected error message")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user