From a8253147379edffedf9d4bcc14bd29bf4a3fd9e5 Mon Sep 17 00:00:00 2001
From: douxu <douxu@clea.com.cn>
Date: Wed, 5 Aug 2026 16:10:47 +0800
Subject: [PATCH] feat(manualsync): add dual-protocol mock server - expose
 manual sync and health-check endpoints on ports 9001 and 9002 - add
 structured request logging and graceful shutdown - cover routing, validation,
 listener cleanup, and shutdown behavior - move async task API documentation
 into docs

---
 .gitignore                        |   4 +
 cmd/manualsyncmock/main.go        |  22 +++
 cmd/manualsyncmock/server.go      | 168 +++++++++++++++++++++
 cmd/manualsyncmock/server_test.go | 234 ++++++++++++++++++++++++++++++
 {doc => docs}/async_task_api.md   |   0
 5 files changed, 428 insertions(+)
 create mode 100644 cmd/manualsyncmock/main.go
 create mode 100644 cmd/manualsyncmock/server.go
 create mode 100644 cmd/manualsyncmock/server_test.go
 rename {doc => docs}/async_task_api.md (100%)

diff --git a/.gitignore b/.gitignore
index 164a18c..c26c054 100644
--- a/.gitignore
+++ b/.gitignore
@@ -30,6 +30,7 @@ go.work
 /configs/**/*.pem
 
 # ai config
+.agents/
 .cursor/
 .claude/
 .codewhale/
@@ -41,3 +42,6 @@ go.work
 ai-debug.log
 *.patch
 *.diff
+docs/agents/
+AGENTS.md
+skills-lock.json
diff --git a/cmd/manualsyncmock/main.go b/cmd/manualsyncmock/main.go
new file mode 100644
index 0000000..e4f186f
--- /dev/null
+++ b/cmd/manualsyncmock/main.go
@@ -0,0 +1,22 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"net"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+
+	err := run(ctx, logger, net.Listen)
+	stop()
+	if err != nil {
+		logger.Error("manual sync mock stopped", "error", err)
+		os.Exit(1)
+	}
+}
diff --git a/cmd/manualsyncmock/server.go b/cmd/manualsyncmock/server.go
new file mode 100644
index 0000000..975e270
--- /dev/null
+++ b/cmd/manualsyncmock/server.go
@@ -0,0 +1,168 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"modelRT/client/manualsync"
+)
+
+type listenFunc func(network, address string) (net.Listener, error)
+
+const (
+	cl3611Address      = ":9001"
+	protocol104Address = ":9002"
+	manualAPIPath      = "/api/manual"
+	healthPath         = "/healthz"
+	maxRequestBody     = 1 << 20
+	shutdownTimeout    = 5 * time.Second
+)
+
+func newProtocolServer(protocol, address string, logger *slog.Logger) *http.Server {
+	mux := http.NewServeMux()
+	mux.HandleFunc("POST "+manualAPIPath, manualHandler(protocol, strings.TrimPrefix(address, ":"), logger))
+	mux.HandleFunc("GET "+healthPath, healthHandler)
+
+	return &http.Server{
+		Addr:              address,
+		Handler:           mux,
+		ReadHeaderTimeout: 5 * time.Second,
+		ReadTimeout:       10 * time.Second,
+		WriteTimeout:      10 * time.Second,
+		IdleTimeout:       60 * time.Second,
+	}
+}
+
+func run(ctx context.Context, logger *slog.Logger, listen listenFunc) error {
+	cl3611Listener, err := listen("tcp", cl3611Address)
+	if err != nil {
+		return fmt.Errorf("listen on %s: %w", cl3611Address, err)
+	}
+	protocol104Listener, err := listen("tcp", protocol104Address)
+	if err != nil {
+		_ = cl3611Listener.Close()
+		return fmt.Errorf("listen on %s: %w", protocol104Address, err)
+	}
+
+	servers := []struct {
+		protocol string
+		server   *http.Server
+		listener net.Listener
+	}{
+		{protocol: "cl3611", server: newProtocolServer("cl3611", cl3611Address, logger), listener: cl3611Listener},
+		{protocol: "104", server: newProtocolServer("104", protocol104Address, logger), listener: protocol104Listener},
+	}
+
+	serveErrors := make(chan error, len(servers))
+	var serversWaitGroup sync.WaitGroup
+	for _, configuredServer := range servers {
+		serversWaitGroup.Add(1)
+		go func() {
+			defer serversWaitGroup.Done()
+			logger.Info("manual sync mock server started",
+				"protocol", configuredServer.protocol,
+				"listen_addr", configuredServer.listener.Addr().String(),
+			)
+			if err := configuredServer.server.Serve(configuredServer.listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+				serveErrors <- fmt.Errorf("%s server failed: %w", configuredServer.protocol, err)
+			}
+		}()
+	}
+
+	var runError error
+	select {
+	case <-ctx.Done():
+	case runError = <-serveErrors:
+	}
+
+	shutdownContext, cancelShutdown := context.WithTimeout(context.Background(), shutdownTimeout)
+	defer cancelShutdown()
+	for _, configuredServer := range servers {
+		if err := configuredServer.server.Shutdown(shutdownContext); err != nil {
+			runError = errors.Join(runError, fmt.Errorf("shut down %s server: %w", configuredServer.protocol, err))
+			_ = configuredServer.server.Close()
+		}
+	}
+	serversWaitGroup.Wait()
+	return runError
+}
+
+func healthHandler(response http.ResponseWriter, _ *http.Request) {
+	response.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	response.WriteHeader(http.StatusOK)
+	_, _ = response.Write([]byte("ok\n"))
+}
+
+func manualHandler(protocol, listenPort string, logger *slog.Logger) http.HandlerFunc {
+	return func(response http.ResponseWriter, request *http.Request) {
+		request.Body = http.MaxBytesReader(response, request.Body, maxRequestBody)
+		var payload manualsync.Request
+		decoder := json.NewDecoder(request.Body)
+		if err := decoder.Decode(&payload); err != nil {
+			logger.LogAttrs(request.Context(), slog.LevelWarn, "invalid manual sync request",
+				slog.String("protocol", protocol),
+				slog.String("listen_port", listenPort),
+				slog.String("remote_addr", request.RemoteAddr),
+				slog.Any("error", err),
+			)
+			var maxBytesError *http.MaxBytesError
+			if errors.As(err, &maxBytesError) {
+				http.Error(response, "request body too large", http.StatusRequestEntityTooLarge)
+				return
+			}
+			http.Error(response, "invalid JSON request body", http.StatusBadRequest)
+			return
+		}
+		var trailing any
+		if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
+			if err == nil {
+				err = errors.New("request body contains multiple JSON documents")
+			}
+			logger.LogAttrs(request.Context(), slog.LevelWarn, "invalid manual sync request",
+				slog.String("protocol", protocol),
+				slog.String("listen_port", listenPort),
+				slog.String("remote_addr", request.RemoteAddr),
+				slog.Any("error", err),
+			)
+			http.Error(response, "invalid JSON request body", http.StatusBadRequest)
+			return
+		}
+
+		logger.LogAttrs(request.Context(), slog.LevelInfo, "manual sync request received",
+			slog.String("protocol", protocol),
+			slog.String("listen_port", listenPort),
+			slog.String("remote_addr", request.RemoteAddr),
+			slog.Int("mode", int(payload.Mode)),
+			slog.Group("target",
+				slog.Int("type", payload.Target.Type),
+				slog.String("station", payload.Target.Station),
+				slog.String("main_pos", payload.Target.MainPos),
+				slog.String("sub_pos", payload.Target.SubPos),
+				slog.String("option", payload.Target.Option),
+			),
+			slog.Attr{Key: "data", Value: samplesLogValue(payload.Data)},
+		)
+		response.WriteHeader(http.StatusNoContent)
+	}
+}
+
+func samplesLogValue(samples []manualsync.SyntheticData) slog.Value {
+	attributes := make([]slog.Attr, 0, len(samples))
+	for index, sample := range samples {
+		attributes = append(attributes, slog.Group(strconv.Itoa(index),
+			slog.Int64("time", sample.Time),
+			slog.Float64("value", sample.Value),
+		))
+	}
+	return slog.GroupValue(attributes...)
+}
diff --git a/cmd/manualsyncmock/server_test.go b/cmd/manualsyncmock/server_test.go
new file mode 100644
index 0000000..0b0df78
--- /dev/null
+++ b/cmd/manualsyncmock/server_test.go
@@ -0,0 +1,234 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestHealthEndpointReportsServerIsReady(t *testing.T) {
+	server := newProtocolServer("cl3611", cl3611Address, slog.Default())
+	request := httptest.NewRequest(http.MethodGet, healthPath, nil)
+	response := httptest.NewRecorder()
+
+	server.Handler.ServeHTTP(response, request)
+
+	result := response.Result()
+	defer result.Body.Close()
+	body, err := io.ReadAll(result.Body)
+	if err != nil {
+		t.Fatalf("read response body: %v", err)
+	}
+	if result.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want %d", result.StatusCode, http.StatusOK)
+	}
+	if contentType := result.Header.Get("Content-Type"); contentType != "text/plain; charset=utf-8" {
+		t.Errorf("Content-Type = %q, want %q", contentType, "text/plain; charset=utf-8")
+	}
+	if string(body) != "ok\n" {
+		t.Errorf("body = %q, want %q", body, "ok\\n")
+	}
+}
+
+func TestManualEndpointLogsStructuredRequest(t *testing.T) {
+	var logs bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logs, nil))
+	server := newProtocolServer("cl3611", cl3611Address, logger)
+	request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{
+		"mode": 1,
+		"data": [{"time": 1736305467506000000, "value": 1.25}],
+		"target": {
+			"type": 1,
+			"station": "001",
+			"main_pos": "ssu001",
+			"sub_pos": "TM1",
+			"option": "RMS"
+		}
+	}`))
+	request.RemoteAddr = "127.0.0.1:52130"
+	response := httptest.NewRecorder()
+
+	server.Handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusNoContent {
+		t.Fatalf("status = %d, want %d", response.Code, http.StatusNoContent)
+	}
+	for _, fragment := range []string{
+		"protocol=cl3611",
+		"listen_port=9001",
+		"remote_addr=127.0.0.1:52130",
+		"mode=1",
+		"target.type=1",
+		"target.station=001",
+		"target.main_pos=ssu001",
+		"target.sub_pos=TM1",
+		"target.option=RMS",
+		"time=1736305467506000000",
+		"value=1.25",
+	} {
+		if !strings.Contains(logs.String(), fragment) {
+			t.Errorf("log %q does not contain %q", logs.String(), fragment)
+		}
+	}
+}
+
+func TestManualEndpointRejectsInvalidJSONAndLogsTheError(t *testing.T) {
+	var logs bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logs, nil))
+	server := newProtocolServer("104", protocol104Address, logger)
+	request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{"mode":`))
+	request.RemoteAddr = "127.0.0.1:52131"
+	response := httptest.NewRecorder()
+
+	server.Handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d", response.Code, http.StatusBadRequest)
+	}
+	for _, fragment := range []string{
+		"level=WARN",
+		"msg=\"invalid manual sync request\"",
+		"protocol=104",
+		"listen_port=9002",
+		"remote_addr=127.0.0.1:52131",
+		"error=",
+	} {
+		if !strings.Contains(logs.String(), fragment) {
+			t.Errorf("log %q does not contain %q", logs.String(), fragment)
+		}
+	}
+}
+
+func TestManualEndpointRejectsOversizedRequest(t *testing.T) {
+	server := newProtocolServer("cl3611", cl3611Address, slog.Default())
+	body := `{"extra":"` + strings.Repeat("a", maxRequestBody) + `"}`
+	request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(body))
+	response := httptest.NewRecorder()
+
+	server.Handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusRequestEntityTooLarge {
+		t.Fatalf("status = %d, want %d", response.Code, http.StatusRequestEntityTooLarge)
+	}
+}
+
+func TestManualEndpointRejectsMultipleJSONDocuments(t *testing.T) {
+	server := newProtocolServer("104", protocol104Address, slog.Default())
+	request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{"mode":1}{"mode":0}`))
+	response := httptest.NewRecorder()
+
+	server.Handler.ServeHTTP(response, request)
+
+	if response.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want %d", response.Code, http.StatusBadRequest)
+	}
+}
+
+func TestRunClosesFirstListenerWhenSecondListenerFails(t *testing.T) {
+	rawListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen on temporary port: %v", err)
+	}
+	firstListener := rawListener.(*net.TCPListener)
+	listenCalls := 0
+	listen := func(_, _ string) (net.Listener, error) {
+		listenCalls++
+		if listenCalls == 1 {
+			return firstListener, nil
+		}
+		return nil, errors.New("port is already in use")
+	}
+
+	err = run(context.Background(), slog.Default(), listen)
+
+	if err == nil || !strings.Contains(err.Error(), protocol104Address) {
+		t.Fatalf("run error = %v, want error containing %q", err, protocol104Address)
+	}
+	if err := firstListener.SetDeadline(time.Now()); err == nil {
+		t.Fatal("first listener is still open")
+	}
+}
+
+func TestRunServesBothProtocolsAndStopsOnCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	addresses := make(chan string, 2)
+	listen := func(_, _ string) (net.Listener, error) {
+		listener, err := net.Listen("tcp", "127.0.0.1:0")
+		if err == nil {
+			addresses <- listener.Addr().String()
+		}
+		return listener, err
+	}
+	runErrors := make(chan error, 1)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	go func() {
+		runErrors <- run(ctx, logger, listen)
+	}()
+
+	cl3611Addr := <-addresses
+	protocol104Addr := <-addresses
+	waitForHealthyEndpoint(t, cl3611Addr)
+	waitForHealthyEndpoint(t, protocol104Addr)
+	cancel()
+
+	select {
+	case err := <-runErrors:
+		if err != nil {
+			t.Fatalf("run returned an error during graceful shutdown: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("servers did not stop after context cancellation")
+	}
+}
+
+func TestProtocolServerEnforcesMethodAndPathRouting(t *testing.T) {
+	server := newProtocolServer("cl3611", cl3611Address, slog.Default())
+	tests := []struct {
+		name       string
+		method     string
+		path       string
+		wantStatus int
+	}{
+		{name: "manual endpoint rejects GET", method: http.MethodGet, path: manualAPIPath, wantStatus: http.StatusMethodNotAllowed},
+		{name: "unknown path is not found", method: http.MethodGet, path: "/unknown", wantStatus: http.StatusNotFound},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			request := httptest.NewRequest(test.method, test.path, nil)
+			response := httptest.NewRecorder()
+
+			server.Handler.ServeHTTP(response, request)
+
+			if response.Code != test.wantStatus {
+				t.Fatalf("status = %d, want %d", response.Code, test.wantStatus)
+			}
+		})
+	}
+}
+
+func waitForHealthyEndpoint(t *testing.T, address string) {
+	t.Helper()
+	client := &http.Client{Timeout: 100 * time.Millisecond}
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		response, err := client.Get("http://" + address + healthPath)
+		if err == nil {
+			_ = response.Body.Close()
+			if response.StatusCode == http.StatusOK {
+				return
+			}
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("health endpoint at %s did not become ready", address)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
diff --git a/doc/async_task_api.md b/docs/async_task_api.md
similarity index 100%
rename from doc/async_task_api.md
rename to docs/async_task_api.md