diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 53cb5410..c8f70777 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -239,6 +239,14 @@ table_owners: - events event_service_principals: - events + event_delivery_attempts: + - events + event_consumer_modes: + - events + event_consumer_checkpoints: + - events + event_reconciliations: + - events final_artifacts: - artifactvault artifact_payload_staging: diff --git a/cmd/events/main.go b/cmd/events/main.go new file mode 100644 index 00000000..9dba1508 --- /dev/null +++ b/cmd/events/main.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "chat-agent-system/internal/config" + "chat-agent-system/internal/db" + "chat-agent-system/internal/events" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func main() { + envPath := flag.String("env", ".env", "环境配置文件路径") + flag.Parse() + + cfg, err := config.Load(*envPath) + if err != nil { + fmt.Fprintf(os.Stderr, "events: load config: %v\n", os.ErrInvalid) + os.Exit(1) + } + port := cfg.HTTPPort + if v := os.Getenv("EVENTS_HTTP_PORT"); v != "" { + if p, perr := parsePort(v); perr == nil { + port = p + } + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + pc, err := pgxpool.ParseConfig(cfg.ContextDSN()) + if err != nil { + fmt.Fprintf(os.Stderr, "events: dsn\n") + os.Exit(1) + } + pc.MaxConns = 8 + pool, err := pgxpool.NewWithConfig(ctx, pc) + if err != nil { + fmt.Fprintf(os.Stderr, "events: pool\n") + os.Exit(1) + } + defer pool.Close() + + if err := db.RunMigrations(ctx, pool, db.ResolveMigrationsDir()); err != nil { + fmt.Fprintf(os.Stderr, "events: migrate\n") + os.Exit(1) + } + + owner := fmt.Sprintf("events-%d", os.Getpid()) + disp := events.NewDispatcher(pool, cfg.Env, owner) + mux := http.NewServeMux() + mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "env": cfg.Env, "owner": owner}) + }) + mux.HandleFunc("GET /ready", func(w http.ResponseWriter, r *http.Request) { + if err := events.PingPool(r.Context(), pool); err != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "DB"}) + return + } + if err := events.SchemaReady(r.Context(), pool); err != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "SCHEMA"}) + return + } + reg := events.NewRegistry() + if err := reg.Register(events.ConsumerNotify); err != nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "REGISTRY"}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"status": "ok"}) + }) + mux.HandleFunc("GET /live", func(w http.ResponseWriter, _ *http.Request) { + if !disp.Alive(30 * time.Second) { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "stale"}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"status": "ok"}) + }) + mux.HandleFunc("GET /metrics", func(w http.ResponseWriter, _ *http.Request) { + m := disp.Metrics + writeJSON(w, http.StatusOK, map[string]any{ + "claimed": m.Claimed.Load(), "acked": m.Acked.Load(), "failed": m.Failed.Load(), + "fenced": m.Fenced.Load(), "dead": m.Dead.Load(), "quarantine": m.Quarantine.Load(), + "heartbeat": m.Heartbeat.Load(), + }) + }) + + srv := &http.Server{Addr: fmt.Sprintf("%s:%d", cfg.HTTPHost, port), Handler: mux} + go func() { + consumers := []string{events.ConsumerNotify, events.ConsumerProjection, events.ConsumerWorkflowAdvance} + _ = disp.Serve(ctx, consumers) + }() + go func() { + <-ctx.Done() + shut, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + disp.StopClaim() + _ = srv.Shutdown(shut) + }() + fmt.Printf("events worker env=%s listen=%s:%d owner=%s mode=cluster-table\n", cfg.Env, cfg.HTTPHost, port, owner) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + os.Exit(1) + } +} + +func parsePort(v string) (int, error) { + var p int + _, err := fmt.Sscanf(v, "%d", &p) + if err != nil || p <= 0 { + return 0, os.ErrInvalid + } + return p, nil +} + +func writeJSON(w http.ResponseWriter, status int, body map[string]any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} diff --git a/deploy/app/Dockerfile b/deploy/app/Dockerfile index 3fa1d126..557e75a7 100644 --- a/deploy/app/Dockerfile +++ b/deploy/app/Dockerfile @@ -18,6 +18,7 @@ COPY go.mod go.sum ./ RUN go mod download COPY . . RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /server ./cmd/server +RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /events ./cmd/events FROM debian:bookworm-slim @@ -28,6 +29,7 @@ RUN set -eux; \ WORKDIR /app COPY --from=go-builder /server /app/server +COPY --from=go-builder /events /app/events COPY --from=web-builder /web/dist /app/web/dist COPY deploy/postgres/goose /app/deploy/postgres/goose # Official markdown must be on disk at /app/templates/docs and /app/configs. diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index fa6c6aa4..3088b47f 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -143,6 +143,35 @@ services: - dev-net - shared-net + events: + build: + context: . + dockerfile: deploy/app/Dockerfile + container_name: shuimu-agent-events-dev + entrypoint: ["/app/events"] + command: ["-env", "/dev/null"] + depends_on: + system-db: + condition: service_healthy + environment: + ENV: dev + HTTP_HOST: "0.0.0.0" + HTTP_PORT: "${EVENTS_HTTP_PORT:-8088}" + EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8088}" + DATA_DIR: /app/data + APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:-dev_secret_db_pass}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_dev}?sslmode=disable + ports: + - "127.0.0.1:${EVENTS_HTTP_PORT:-8088}:${EVENTS_HTTP_PORT:-8088}" + healthcheck: + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8088}/health"] + interval: 10s + timeout: 5s + retries: 6 + start_period: 15s + restart: unless-stopped + networks: + - dev-net + networks: dev-net: name: chat-agent-dev-net diff --git a/docker-compose.main.yml b/docker-compose.main.yml index 553c770d..8de3feb0 100644 --- a/docker-compose.main.yml +++ b/docker-compose.main.yml @@ -142,6 +142,35 @@ services: - main-net - shared-net + events: + build: + context: . + dockerfile: deploy/app/Dockerfile + container_name: shuimu-agent-events-main + entrypoint: ["/app/events"] + command: ["-env", "/dev/null"] + depends_on: + system-db: + condition: service_healthy + environment: + ENV: main + HTTP_HOST: "0.0.0.0" + HTTP_PORT: "${EVENTS_HTTP_PORT:-8090}" + EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8090}" + DATA_DIR: /app/data + APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:?SYSTEM_DB_PASSWORD is required for main/production}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_main}?sslmode=disable + ports: + - "127.0.0.1:${EVENTS_HTTP_PORT:-8090}:${EVENTS_HTTP_PORT:-8090}" + healthcheck: + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8090}/health"] + interval: 10s + timeout: 5s + retries: 6 + start_period: 15s + restart: always + networks: + - main-net + networks: main-net: name: chat-agent-main-net diff --git a/docker-compose.test.yml b/docker-compose.test.yml index d0d880e4..aecb168f 100644 --- a/docker-compose.test.yml +++ b/docker-compose.test.yml @@ -141,6 +141,35 @@ services: - test-net - shared-net + events: + build: + context: . + dockerfile: deploy/app/Dockerfile + container_name: shuimu-agent-events-test + entrypoint: ["/app/events"] + command: ["-env", "/dev/null"] + depends_on: + system-db: + condition: service_healthy + environment: + ENV: test + HTTP_HOST: "0.0.0.0" + HTTP_PORT: "${EVENTS_HTTP_PORT:-8089}" + EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8089}" + DATA_DIR: /app/data + APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:-test_secret_db_pass}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_test}?sslmode=disable + ports: + - "127.0.0.1:${EVENTS_HTTP_PORT:-8089}:${EVENTS_HTTP_PORT:-8089}" + healthcheck: + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8089}/health"] + interval: 10s + timeout: 5s + retries: 6 + start_period: 15s + restart: unless-stopped + networks: + - test-net + networks: test-net: name: chat-agent-test-net