feat: add msglib sidecar go client
This commit is contained in:
zhaoyilun
231
internal/msglib/sidecar_client.go
Normal file
231
internal/msglib/sidecar_client.go
Normal file
@@ -0,0 +1,231 @@
|
||||
package msglib
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
Protocol = "isphere.msglib.v1"
|
||||
DefaultPassword = "123"
|
||||
)
|
||||
|
||||
var requestCounter uint64
|
||||
|
||||
// Config describes the local sidecar process and copied MsgLib DB inputs.
|
||||
type Config struct {
|
||||
SidecarPath string
|
||||
SidecarArgs []string
|
||||
SQLiteDLLPath string
|
||||
DBPath string
|
||||
Password string
|
||||
Timeout time.Duration
|
||||
ExtraEnv []string
|
||||
}
|
||||
|
||||
// ConfigFromEnv reads the C28 environment shape without wiring it into MCP tools.
|
||||
func ConfigFromEnv() (Config, error) {
|
||||
cfg := Config{
|
||||
SidecarPath: strings.TrimSpace(os.Getenv("ISPHERE_MSGLIB_SIDECAR_EXE")),
|
||||
SQLiteDLLPath: strings.TrimSpace(os.Getenv("ISPHERE_MSGLIB_SQLITE_DLL")),
|
||||
DBPath: strings.TrimSpace(os.Getenv("ISPHERE_MSGLIB_DB")),
|
||||
Password: strings.TrimSpace(os.Getenv("ISPHERE_MSGLIB_PASSWORD")),
|
||||
}
|
||||
if cfg.Password == "" {
|
||||
cfg.Password = DefaultPassword
|
||||
}
|
||||
if cfg.SidecarPath == "" {
|
||||
return cfg, errors.New("ISPHERE_MSGLIB_SIDECAR_EXE is not configured")
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
// NewClient returns a process-backed MsgLib sidecar client.
|
||||
func NewClient(cfg Config) *Client {
|
||||
if cfg.Password == "" {
|
||||
cfg.Password = DefaultPassword
|
||||
}
|
||||
if cfg.Timeout == 0 {
|
||||
cfg.Timeout = 10 * time.Second
|
||||
}
|
||||
return &Client{config: cfg}
|
||||
}
|
||||
|
||||
type Client struct {
|
||||
config Config
|
||||
}
|
||||
|
||||
type SelfCheckResult struct {
|
||||
SidecarName string `json:"sidecar_name"`
|
||||
SidecarVersion string `json:"sidecar_version"`
|
||||
Protocol string `json:"protocol"`
|
||||
Runtime string `json:"runtime"`
|
||||
ProcessBits int `json:"process_bits"`
|
||||
ReadOnly bool `json:"read_only"`
|
||||
MutatesDB bool `json:"mutates_db"`
|
||||
}
|
||||
|
||||
type DisplaySource struct {
|
||||
Source string `json:"source"`
|
||||
Table string `json:"table"`
|
||||
Available bool `json:"available"`
|
||||
RequiredColumns []string `json:"required_columns"`
|
||||
PresentColumns []string `json:"present_columns"`
|
||||
}
|
||||
|
||||
type SidecarError struct {
|
||||
Code string `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
func (e *SidecarError) Error() string {
|
||||
if e == nil {
|
||||
return "msglib sidecar error"
|
||||
}
|
||||
if e.Code == "" {
|
||||
return "msglib sidecar error: " + e.Message
|
||||
}
|
||||
if e.Message == "" {
|
||||
return "msglib sidecar error: " + e.Code
|
||||
}
|
||||
return fmt.Sprintf("msglib sidecar error %s: %s", e.Code, e.Message)
|
||||
}
|
||||
|
||||
func (c *Client) SelfCheck(ctx context.Context) (SelfCheckResult, error) {
|
||||
var out SelfCheckResult
|
||||
err := c.call(ctx, "self_check", map[string]any{}, &out)
|
||||
return out, err
|
||||
}
|
||||
|
||||
func (c *Client) DisplaySources(ctx context.Context) ([]DisplaySource, error) {
|
||||
args := map[string]any{
|
||||
"sqlite_dll_path": c.config.SQLiteDLLPath,
|
||||
"db_path": c.config.DBPath,
|
||||
"password": c.config.Password,
|
||||
}
|
||||
var out displaySourcesData
|
||||
if err := c.call(ctx, "display_sources", args, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out.DisplaySources, nil
|
||||
}
|
||||
|
||||
type displaySourcesData struct {
|
||||
DisplaySources []DisplaySource `json:"display_sources"`
|
||||
}
|
||||
|
||||
type requestEnvelope struct {
|
||||
Protocol string `json:"protocol"`
|
||||
RequestID string `json:"request_id"`
|
||||
Op string `json:"op"`
|
||||
Args map[string]any `json:"args,omitempty"`
|
||||
}
|
||||
|
||||
type responseEnvelope struct {
|
||||
Protocol string `json:"protocol"`
|
||||
RequestID string `json:"request_id"`
|
||||
Op string `json:"op"`
|
||||
OK bool `json:"ok"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
Error *SidecarError `json:"error"`
|
||||
Warnings []string `json:"warnings"`
|
||||
}
|
||||
|
||||
func (c *Client) call(ctx context.Context, op string, args map[string]any, out any) error {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if strings.TrimSpace(c.config.SidecarPath) == "" {
|
||||
return errors.New("msglib sidecar path is not configured")
|
||||
}
|
||||
|
||||
requestID := nextRequestID()
|
||||
req := requestEnvelope{
|
||||
Protocol: Protocol,
|
||||
RequestID: requestID,
|
||||
Op: op,
|
||||
Args: args,
|
||||
}
|
||||
payload, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal msglib sidecar request: %w", err)
|
||||
}
|
||||
payload = append(payload, '\n')
|
||||
|
||||
callCtx := ctx
|
||||
var cancel context.CancelFunc
|
||||
if c.config.Timeout > 0 {
|
||||
callCtx, cancel = context.WithTimeout(ctx, c.config.Timeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
cmd := exec.CommandContext(callCtx, c.config.SidecarPath, c.config.SidecarArgs...)
|
||||
cmd.Stdin = bytes.NewReader(payload)
|
||||
var stdout bytes.Buffer
|
||||
var stderr bytes.Buffer
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
if len(c.config.ExtraEnv) > 0 {
|
||||
cmd.Env = append(os.Environ(), c.config.ExtraEnv...)
|
||||
}
|
||||
|
||||
runErr := cmd.Run()
|
||||
if callCtx.Err() != nil {
|
||||
if errors.Is(callCtx.Err(), context.DeadlineExceeded) {
|
||||
return fmt.Errorf("msglib sidecar %s timed out: %w", op, context.DeadlineExceeded)
|
||||
}
|
||||
return fmt.Errorf("msglib sidecar %s context failed: %w", op, callCtx.Err())
|
||||
}
|
||||
if runErr != nil {
|
||||
stderrText := strings.TrimSpace(stderr.String())
|
||||
if stderrText == "" {
|
||||
return fmt.Errorf("msglib sidecar %s process failed: %w", op, runErr)
|
||||
}
|
||||
return fmt.Errorf("msglib sidecar %s process failed: %w: %s", op, runErr, stderrText)
|
||||
}
|
||||
|
||||
body := bytes.TrimSpace(stdout.Bytes())
|
||||
if len(body) == 0 {
|
||||
return fmt.Errorf("msglib sidecar %s returned empty stdout", op)
|
||||
}
|
||||
|
||||
var resp responseEnvelope
|
||||
if err := json.Unmarshal(body, &resp); err != nil {
|
||||
return fmt.Errorf("decode msglib sidecar %s response: %w", op, err)
|
||||
}
|
||||
if resp.Protocol != Protocol {
|
||||
return fmt.Errorf("msglib sidecar %s response protocol = %q, want %q", op, resp.Protocol, Protocol)
|
||||
}
|
||||
if resp.RequestID != requestID {
|
||||
return fmt.Errorf("msglib sidecar %s response request_id = %q, want %q", op, resp.RequestID, requestID)
|
||||
}
|
||||
if resp.Op != op {
|
||||
return fmt.Errorf("msglib sidecar response op = %q, want %q", resp.Op, op)
|
||||
}
|
||||
if !resp.OK {
|
||||
if resp.Error != nil {
|
||||
return fmt.Errorf("msglib sidecar %s failed: %w", op, resp.Error)
|
||||
}
|
||||
return fmt.Errorf("msglib sidecar %s failed", op)
|
||||
}
|
||||
if out == nil || len(resp.Data) == 0 || bytes.Equal(resp.Data, []byte("null")) {
|
||||
return nil
|
||||
}
|
||||
if err := json.Unmarshal(resp.Data, out); err != nil {
|
||||
return fmt.Errorf("decode msglib sidecar %s data: %w", op, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func nextRequestID() string {
|
||||
n := atomic.AddUint64(&requestCounter, 1)
|
||||
return fmt.Sprintf("go-%d-%d", time.Now().UnixNano(), n)
|
||||
}
|
||||
217
internal/msglib/sidecar_client_test.go
Normal file
217
internal/msglib/sidecar_client_test.go
Normal file
@@ -0,0 +1,217 @@
|
||||
package msglib
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestSelfCheckSendsProtocolEnvelopeAndDecodesResponse(t *testing.T) {
|
||||
client := newFakeClient(t, "self_check_ok", Config{})
|
||||
|
||||
got, err := client.SelfCheck(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("SelfCheck returned error: %v", err)
|
||||
}
|
||||
if got.SidecarName != "MsgLibReadSidecar" {
|
||||
t.Fatalf("SidecarName = %q, want MsgLibReadSidecar", got.SidecarName)
|
||||
}
|
||||
if got.Protocol != Protocol {
|
||||
t.Fatalf("Protocol = %q, want %q", got.Protocol, Protocol)
|
||||
}
|
||||
if got.ProcessBits != 32 || !got.ReadOnly || got.MutatesDB {
|
||||
t.Fatalf("unexpected safety fields: bits=%d read_only=%v mutates_db=%v", got.ProcessBits, got.ReadOnly, got.MutatesDB)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDisplaySourcesPassesConfiguredPathsAndDecodesMetadata(t *testing.T) {
|
||||
cfg := Config{
|
||||
SQLiteDLLPath: `C:\fixture\System.Data.SQLite.dll`,
|
||||
DBPath: `C:\fixture\MsgLib.db`,
|
||||
Password: "123",
|
||||
}
|
||||
client := newFakeClient(t, "display_sources_ok", cfg)
|
||||
|
||||
got, err := client.DisplaySources(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("DisplaySources returned error: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("len(DisplaySources) = %d, want 2", len(got))
|
||||
}
|
||||
if got[0].Source != "contacts_roster" || got[0].Table != "TD_Roster" || !got[0].Available {
|
||||
t.Fatalf("unexpected first display source: %+v", got[0])
|
||||
}
|
||||
if strings.Join(got[0].RequiredColumns, ",") != "UserJID,Name" {
|
||||
t.Fatalf("required columns = %#v", got[0].RequiredColumns)
|
||||
}
|
||||
if got[1].Source != "work_group_auth" || got[1].Available {
|
||||
t.Fatalf("unexpected second display source: %+v", got[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientReturnsStructuredSidecarError(t *testing.T) {
|
||||
client := newFakeClient(t, "sidecar_error", Config{})
|
||||
|
||||
_, err := client.SelfCheck(context.Background())
|
||||
if err == nil {
|
||||
t.Fatal("SelfCheck returned nil error, want sidecar error")
|
||||
}
|
||||
var sidecarErr *SidecarError
|
||||
if !errors.As(err, &sidecarErr) {
|
||||
t.Fatalf("error %T %[1]v does not wrap *SidecarError", err)
|
||||
}
|
||||
if sidecarErr.Code != "DB_OPEN_FAILED" || !strings.Contains(sidecarErr.Message, "read-only open failed") {
|
||||
t.Fatalf("unexpected sidecar error: %+v", sidecarErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientTimeoutStopsSidecar(t *testing.T) {
|
||||
client := newFakeClient(t, "hang", Config{Timeout: 150 * time.Millisecond})
|
||||
|
||||
_, err := client.SelfCheck(context.Background())
|
||||
if err == nil {
|
||||
t.Fatal("SelfCheck returned nil error, want timeout")
|
||||
}
|
||||
if !errors.Is(err, context.DeadlineExceeded) && !strings.Contains(strings.ToLower(err.Error()), "deadline") {
|
||||
t.Fatalf("SelfCheck error = %v, want deadline exceeded", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigFromEnvReadsSidecarShape(t *testing.T) {
|
||||
t.Setenv("ISPHERE_MSGLIB_SIDECAR_EXE", `C:\tools\MsgLibReadSidecar.exe`)
|
||||
t.Setenv("ISPHERE_MSGLIB_SQLITE_DLL", `C:\tools\System.Data.SQLite.dll`)
|
||||
t.Setenv("ISPHERE_MSGLIB_DB", `C:\copy\MsgLib.db`)
|
||||
|
||||
cfg, err := ConfigFromEnv()
|
||||
if err != nil {
|
||||
t.Fatalf("ConfigFromEnv returned error: %v", err)
|
||||
}
|
||||
if cfg.SidecarPath != `C:\tools\MsgLibReadSidecar.exe` {
|
||||
t.Fatalf("SidecarPath = %q", cfg.SidecarPath)
|
||||
}
|
||||
if cfg.SQLiteDLLPath != `C:\tools\System.Data.SQLite.dll` || cfg.DBPath != `C:\copy\MsgLib.db` {
|
||||
t.Fatalf("unexpected DB config: %+v", cfg)
|
||||
}
|
||||
if cfg.Password != "123" {
|
||||
t.Fatalf("Password = %q, want default 123", cfg.Password)
|
||||
}
|
||||
}
|
||||
|
||||
func newFakeClient(t *testing.T, mode string, overrides Config) *Client {
|
||||
t.Helper()
|
||||
exe, err := os.Executable()
|
||||
if err != nil {
|
||||
t.Fatalf("os.Executable: %v", err)
|
||||
}
|
||||
cfg := overrides
|
||||
cfg.SidecarPath = exe
|
||||
cfg.SidecarArgs = []string{"-test.run=TestFakeSidecarProcess", "--"}
|
||||
cfg.ExtraEnv = append(cfg.ExtraEnv, "ISPHERE_MSGLIB_FAKE_SIDECAR=1", "ISPHERE_MSGLIB_FAKE_MODE="+mode)
|
||||
if cfg.Timeout == 0 {
|
||||
cfg.Timeout = 2 * time.Second
|
||||
}
|
||||
return NewClient(cfg)
|
||||
}
|
||||
|
||||
func TestFakeSidecarProcess(t *testing.T) {
|
||||
if os.Getenv("ISPHERE_MSGLIB_FAKE_SIDECAR") != "1" {
|
||||
return
|
||||
}
|
||||
os.Exit(fakeSidecarMain())
|
||||
}
|
||||
|
||||
func fakeSidecarMain() int {
|
||||
mode := os.Getenv("ISPHERE_MSGLIB_FAKE_MODE")
|
||||
if mode == "hang" {
|
||||
time.Sleep(5 * time.Second)
|
||||
return 0
|
||||
}
|
||||
|
||||
var req map[string]any
|
||||
if err := json.NewDecoder(os.Stdin).Decode(&req); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "decode request: %v\n", err)
|
||||
return 2
|
||||
}
|
||||
protocol, _ := req["protocol"].(string)
|
||||
op, _ := req["op"].(string)
|
||||
requestID, _ := req["request_id"].(string)
|
||||
if protocol != Protocol || op == "" || requestID == "" {
|
||||
fmt.Fprintf(os.Stderr, "bad envelope: protocol=%q op=%q request_id=%q\n", protocol, op, requestID)
|
||||
return 2
|
||||
}
|
||||
|
||||
if mode == "sidecar_error" {
|
||||
return writeFakeResponse(map[string]any{
|
||||
"protocol": Protocol,
|
||||
"request_id": requestID,
|
||||
"op": op,
|
||||
"ok": false,
|
||||
"data": nil,
|
||||
"error": map[string]any{
|
||||
"code": "DB_OPEN_FAILED",
|
||||
"message": "read-only open failed",
|
||||
},
|
||||
"warnings": []any{},
|
||||
})
|
||||
}
|
||||
|
||||
switch op {
|
||||
case "self_check":
|
||||
return writeFakeResponse(map[string]any{
|
||||
"protocol": Protocol,
|
||||
"request_id": requestID,
|
||||
"op": op,
|
||||
"ok": true,
|
||||
"data": map[string]any{
|
||||
"sidecar_name": "MsgLibReadSidecar",
|
||||
"sidecar_version": "0.1.0",
|
||||
"protocol": Protocol,
|
||||
"process_bits": 32,
|
||||
"read_only": true,
|
||||
"mutates_db": false,
|
||||
},
|
||||
"error": nil,
|
||||
"warnings": []any{},
|
||||
})
|
||||
case "display_sources":
|
||||
args, _ := req["args"].(map[string]any)
|
||||
if args["sqlite_dll_path"] != `C:\fixture\System.Data.SQLite.dll` || args["db_path"] != `C:\fixture\MsgLib.db` || args["password"] != "123" {
|
||||
fmt.Fprintf(os.Stderr, "bad display_sources args: %#v\n", args)
|
||||
return 2
|
||||
}
|
||||
return writeFakeResponse(map[string]any{
|
||||
"protocol": Protocol,
|
||||
"request_id": requestID,
|
||||
"op": op,
|
||||
"ok": true,
|
||||
"data": map[string]any{
|
||||
"metadata_only": true,
|
||||
"read_only": true,
|
||||
"message_body_values_returned": false,
|
||||
"display_sources": []any{
|
||||
map[string]any{"source": "contacts_roster", "table": "TD_Roster", "available": true, "required_columns": []any{"UserJID", "Name"}, "present_columns": []any{"UserJID", "Name", "Mobile"}},
|
||||
map[string]any{"source": "work_group_auth", "table": "TD_WorkGroupAuth", "available": false, "required_columns": []any{"GroupJID", "GroupName"}, "present_columns": []any{"GroupJID"}},
|
||||
},
|
||||
},
|
||||
"error": nil,
|
||||
"warnings": []any{},
|
||||
})
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "unexpected op: %s\n", op)
|
||||
return 2
|
||||
}
|
||||
}
|
||||
|
||||
func writeFakeResponse(resp map[string]any) int {
|
||||
if err := json.NewEncoder(os.Stdout).Encode(resp); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "encode response: %v\n", err)
|
||||
return 2
|
||||
}
|
||||
return 0
|
||||
}
|
||||
Reference in New Issue
Block a user