refactor: rename iot client to stateless gateway
This commit is contained in:
@@ -6,7 +6,7 @@
|
||||
| --- | --- |
|
||||
| `api` | Gin HTTP API / BFF、同步事务、JWT 鉴权与统一响应 |
|
||||
| `worker` | Outbox 投递、超时扫描与可重试异步任务 |
|
||||
| `iot-client` | 管理系统/App 的设备上下行 HTTP 边界、幂等受理与状态查询 |
|
||||
| `iot-gateway` | API/Worker 与 IoT Server 之间的无状态上下行 HTTP 轻网关 |
|
||||
| `iot-server` | 内嵌 Mochi MQTT Broker、厂商二进制协议、遥测、命令与回执 |
|
||||
|
||||
四个进程沿用统一的工程机制;API 是命令、Outbox 和上行报文的数据库事实主责,Worker 负责投递,IoT Client 负责系统接口,IoT Server 负责设备协议。
|
||||
|
||||
@@ -50,4 +50,3 @@ Payment:
|
||||
|
||||
IoT:
|
||||
InternalServiceToken: change-me-iot-internal-token
|
||||
ClientBaseURL: http://127.0.0.1:12429
|
||||
|
||||
@@ -66,10 +66,9 @@ type PaymentConfig struct {
|
||||
Wechat WechatPayConfig `yaml:"Wechat"`
|
||||
}
|
||||
|
||||
// IoTConfig 保存 API、Worker 与 IoT Client 间的内部认证和地址。
|
||||
// IoTConfig 保存 API 与 IoT 链路间的内部认证配置。
|
||||
type IoTConfig struct {
|
||||
InternalServiceToken string `yaml:"InternalServiceToken"`
|
||||
ClientBaseURL string `yaml:"ClientBaseURL"`
|
||||
}
|
||||
|
||||
// SrvConfig 与仓库现有进程的配置结构保持一致。
|
||||
@@ -110,8 +109,8 @@ func New(srvKey string) {
|
||||
if Spec.Payment.ExpireMinutes <= 0 || Spec.Payment.RefundWindowDays <= 0 {
|
||||
panic("Payment expiration and refund window must be greater than zero")
|
||||
}
|
||||
if len(strings.TrimSpace(Spec.IoT.InternalServiceToken)) < 16 || !strings.HasPrefix(Spec.IoT.ClientBaseURL, "http") {
|
||||
panic("IoT internal token and client base URL are required")
|
||||
if len(strings.TrimSpace(Spec.IoT.InternalServiceToken)) < 16 {
|
||||
panic("IoT internal token is required")
|
||||
}
|
||||
registerURL, err := url.ParseRequestURI(Spec.Global.UserRegisterURL)
|
||||
if err != nil || (registerURL.Scheme != "http" && registerURL.Scheme != "https") || registerURL.Host == "" {
|
||||
|
||||
@@ -3,6 +3,6 @@ go 1.26.1
|
||||
use (
|
||||
./api
|
||||
./worker
|
||||
./iot-client
|
||||
./iot-gateway
|
||||
./iot-server
|
||||
)
|
||||
|
||||
@@ -2,15 +2,26 @@ cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4=
|
||||
cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
|
||||
cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs=
|
||||
cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10=
|
||||
github.com/DataDog/zstd v1.4.5/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
|
||||
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.31.0 h1:DHa2U07rk8syqvCge0QIGMCE1WxGj9njT44GH7zNJLQ=
|
||||
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.31.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0=
|
||||
github.com/agiledragon/gomonkey v2.0.2+incompatible h1:eXKi9/piiC3cjJD1658mEE2o3NjkJ5vDLgYjCQu0Xlw=
|
||||
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
|
||||
github.com/alicebob/miniredis/v2 v2.23.0/go.mod h1:XNqvJdQJv5mSuVMc0ynneafpnL/zv52acZ6kqeS0t88=
|
||||
github.com/antihax/optional v1.0.0 h1:xK2lYat7ZLaVVcIuj82J8kIro4V6kDe0AUDFboUCwcg=
|
||||
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
|
||||
github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os8IaYg++6uMOdKK83QtkkvJik=
|
||||
github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4=
|
||||
github.com/cockroachdb/errors v1.11.1/go.mod h1:8MUxA3Gi6b25tYlFEBGLf+D8aISL+M4MIpiWMSNRfxw=
|
||||
github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs=
|
||||
github.com/cockroachdb/pebble v1.1.0/go.mod h1:sEHm5NOXxyiAoKWhoFxT8xMgd/f3RA6qUqQ1BXKrh2E=
|
||||
github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg=
|
||||
github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06/go.mod h1:7nc4anLGjupUW/PeY5qiNYsdNXj7zopG+eqsS7To5IQ=
|
||||
github.com/dgraph-io/badger/v4 v4.2.0/go.mod h1:qfCqhPoWDFJRx1gp5QwwyGo8xk1lbHUxvK9nK0OGAak=
|
||||
github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkzgwUve0VDWWA=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/elastic/elastic-transport-go/v8 v8.9.0 h1:KeT/2P54F0xS0S8Y3Pf+tFDg4HmBgReQMB+BMz8dDAs=
|
||||
@@ -25,18 +36,24 @@ github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an
|
||||
github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4=
|
||||
github.com/envoyproxy/protoc-gen-validate v1.3.3 h1:MVQghNeW+LZcmXe7SY1V36Z+WFMDjpqGAGacLe2T0ds=
|
||||
github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0=
|
||||
github.com/getsentry/sentry-go v0.18.0/go.mod h1:Kgon4Mby+FJ7ZWHFUAZgVaIa8sxHtnRJRLTXZr51aKQ=
|
||||
github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
|
||||
github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
|
||||
github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
|
||||
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
|
||||
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
|
||||
github.com/godbus/dbus/v5 v5.0.4 h1:9349emZab16e7zQvpmsbtjc18ykshndd8y2PG3sgJbA=
|
||||
github.com/golang/glog v1.2.5 h1:DrW6hGnjIhtvhOIiAKT6Psh/Kd/ldepEa81DKeiRJ5I=
|
||||
github.com/golang/glog v1.2.5/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwmO+w=
|
||||
github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
|
||||
github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
|
||||
github.com/google/flatbuffers v1.12.1/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
|
||||
github.com/google/gofuzz v1.0.0 h1:A8PeW59pxE9IoFRqBp37U+mSNaQoZ46F1f0f863XSXw=
|
||||
github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1 h1:qnpSQwGEnkcRpTqNOIR6bJbR0gAorgP9CSALpRcKoAA=
|
||||
github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1/go.mod h1:lXGCsh6c22WGtjr+qGHj1otzZpV/1kwTMAqkwZsnWRU=
|
||||
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 h1:pRhl55Yx1eC7BZ1N+BBWwnKaMyD8uC+34TLdndZMAKk=
|
||||
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0/go.mod h1:XKMd7iuf/RGPSMJ/U4HP0zS2Z9Fh8Ps9a+6X26m/tmI=
|
||||
github.com/jinzhu/copier v0.3.5/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg=
|
||||
github.com/jordanlewis/gcassert v0.0.0-20250430164644-389ef753e22e h1:a+PGEeXb+exwBS3NboqXHyxarD9kaboBbrSp+7GuBuc=
|
||||
github.com/jordanlewis/gcassert v0.0.0-20250430164644-389ef753e22e/go.mod h1:ZybsQk6DWyN5t7An1MuPm1gtSZ1xDaTXS9ZjIOxvQrk=
|
||||
github.com/kisielk/errcheck v1.5.0 h1:e8esj/e4R+SAOwFwN+n3zr0nYeCyeweozKfO23MvHzY=
|
||||
@@ -46,6 +63,7 @@ github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBF
|
||||
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU=
|
||||
github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc=
|
||||
@@ -55,6 +73,7 @@ github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZd
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30 h1:BHT1/DKsYDGkUgQ2jmMaozVcdk+sVfz0+1ZJq4zkWgw=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
|
||||
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
|
||||
github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
|
||||
@@ -67,8 +86,6 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg
|
||||
github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
|
||||
github.com/rogpeppe/fastuuid v1.2.0 h1:Ppwyp6VYCF1nvBTXL3trRso7mXMlRrw9ooo375wvi2s=
|
||||
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
|
||||
github.com/rs/xid v1.4.0 h1:qd7wPTDkN6KQx2VmMBLrpHkiyQwgFXRnkOLacUiaSNY=
|
||||
github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
|
||||
github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI=
|
||||
github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
|
||||
github.com/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo=
|
||||
@@ -87,12 +104,16 @@ github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gi
|
||||
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM=
|
||||
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI=
|
||||
github.com/yuin/goldmark v1.2.1 h1:ruQGxdhGHe7FWOJPT0mKs5+pD2Xs1Bm/kdGlHO04FmM=
|
||||
github.com/yuin/gopher-lua v0.0.0-20210529063254-f4c35e4016d9/go.mod h1:E1AXubJBdNmFERAOucpDIxNzeGfLzg0mYh+UfMWdChA=
|
||||
github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
|
||||
github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
|
||||
go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
|
||||
go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
|
||||
go.opentelemetry.io/contrib/detectors/gcp v1.42.0 h1:kpt2PEJuOuqYkPcktfJqWWDjTEd/FNgrxcniL7kQrXQ=
|
||||
go.opentelemetry.io/contrib/detectors/gcp v1.42.0/go.mod h1:W9zQ439utxymRrXsUOzZbFX4JhLxXU4+ZnCt8GG7yA8=
|
||||
go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
|
||||
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
|
||||
golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc=
|
||||
golang.org/x/mod v0.34.0 h1:xIHgNUUnW6sYkcM5Jleh05DvLOtwc6RitGHbDk4akRI=
|
||||
golang.org/x/mod v0.34.0/go.mod h1:ykgH52iCZe79kzLLMhyCUzhMci+nQj+0XkbXpNYtVjY=
|
||||
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
# IoT Client
|
||||
|
||||
管理系统与 App 的上下行设备接口边界。外部业务请求应先经过 Platform API 的 JWT、角色、对象归属和安全状态校验;API/Worker 使用内部令牌调用本服务。命令要求 `identity`、`idempotency_key`、过期时间,并返回可查询状态,不能把 HTTP 受理视为设备执行成功。
|
||||
|
||||
设备上行报文由 IoT Server 回调本服务,再转交 Platform API 持久化和审计。
|
||||
@@ -1,5 +0,0 @@
|
||||
module git.apinb.com/heqiapp/platforms/backend/iot-client
|
||||
|
||||
go 1.26.1
|
||||
|
||||
require gopkg.in/yaml.v3 v3.0.1
|
||||
5
backend/iot-gateway/README.md
Normal file
5
backend/iot-gateway/README.md
Normal file
@@ -0,0 +1,5 @@
|
||||
# IoT Gateway
|
||||
|
||||
API/Worker 与 IoT Server 之间的无状态轻网关。外部业务请求必须先经过 Platform API 的 JWT、角色、对象归属和安全状态校验;Worker 使用内部令牌经本服务下发命令。网关不保存命令、幂等或设备事实,只透传 `identity`、`idempotency_key` 和过期时间,并返回 IoT Server 的派发结果。
|
||||
|
||||
设备上行报文由 IoT Server 回调本服务,再转交 Platform API 持久化和审计;命令与幂等事实、最终状态查询统一由 API/PostgreSQL 提供。
|
||||
@@ -3,15 +3,15 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-client/internal/config"
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-client/internal/service"
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-gateway/internal/config"
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-gateway/internal/service"
|
||||
"log"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
func main() {
|
||||
path := flag.String("config", "etc/platform_iot_client_dev.yaml", "YAML 配置文件")
|
||||
path := flag.String("config", "etc/platform_iot_gateway_dev.yaml", "YAML 配置文件")
|
||||
flag.Parse()
|
||||
cfg, err := config.Load(*path)
|
||||
if err != nil {
|
||||
@@ -1,4 +1,4 @@
|
||||
Service: platform-iot-client
|
||||
Service: platform-iot-gateway
|
||||
HTTP:
|
||||
Address: 127.0.0.1:12429
|
||||
InternalToken: change-me-iot-internal-token
|
||||
5
backend/iot-gateway/go.mod
Normal file
5
backend/iot-gateway/go.mod
Normal file
@@ -0,0 +1,5 @@
|
||||
module git.apinb.com/heqiapp/platforms/backend/iot-gateway
|
||||
|
||||
go 1.26.1
|
||||
|
||||
require gopkg.in/yaml.v3 v3.0.1
|
||||
@@ -9,20 +9,15 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-client/internal/config"
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-gateway/internal/config"
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
cfg config.Config
|
||||
client *http.Client
|
||||
mu sync.RWMutex
|
||||
byIdentity map[string]Command
|
||||
byIdempotency map[string]string
|
||||
http *http.Server
|
||||
cfg config.Config
|
||||
client *http.Client
|
||||
http *http.Server
|
||||
}
|
||||
type Command struct {
|
||||
Identity string `json:"identity"`
|
||||
@@ -37,11 +32,6 @@ type Command struct {
|
||||
Controller byte `json:"controller"`
|
||||
Loop byte `json:"loop"`
|
||||
Component byte `json:"component"`
|
||||
Status string `json:"status"`
|
||||
PacketNumber uint16 `json:"packet_number,omitempty"`
|
||||
ErrorCode string `json:"error_code,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
type deviceEnvelope struct {
|
||||
Type, Topic, DeviceID, ReceivedAt, PayloadHex string
|
||||
@@ -49,13 +39,12 @@ type deviceEnvelope struct {
|
||||
}
|
||||
|
||||
func New(cfg config.Config) *Service {
|
||||
return &Service{cfg: cfg, client: &http.Client{Timeout: 12 * time.Second}, byIdentity: map[string]Command{}, byIdempotency: map[string]string{}}
|
||||
return &Service{cfg: cfg, client: &http.Client{Timeout: 12 * time.Second}}
|
||||
}
|
||||
func (s *Service) Run(ctx context.Context) error {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
|
||||
mux.HandleFunc("/v1/device-commands", s.commands)
|
||||
mux.HandleFunc("/v1/device-commands/", s.commandStatus)
|
||||
mux.HandleFunc("/internal/v1/device-messages", s.deviceMessage)
|
||||
s.http = &http.Server{Addr: s.cfg.HTTP.Address, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
|
||||
go func() {
|
||||
@@ -95,56 +84,12 @@ func (s *Service) commands(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "command expired", 422)
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
if existingID, ok := s.byIdempotency[cmd.IdempotencyKey]; ok {
|
||||
existing := s.byIdentity[existingID]
|
||||
s.mu.Unlock()
|
||||
writeJSON(w, http.StatusOK, existing)
|
||||
return
|
||||
}
|
||||
cmd.Status = "accepted"
|
||||
cmd.CreatedAt = time.Now().UTC()
|
||||
cmd.UpdatedAt = cmd.CreatedAt
|
||||
s.byIdentity[cmd.Identity] = cmd
|
||||
s.byIdempotency[cmd.IdempotencyKey] = cmd.Identity
|
||||
s.mu.Unlock()
|
||||
status, packet, err := s.dispatch(r.Context(), cmd)
|
||||
s.mu.Lock()
|
||||
current := s.byIdentity[cmd.Identity]
|
||||
current.UpdatedAt = time.Now().UTC()
|
||||
if err != nil {
|
||||
current.Status = "dispatch_failed"
|
||||
current.ErrorCode = "IOT_DISPATCH_FAILED"
|
||||
} else {
|
||||
current.Status = status
|
||||
current.PacketNumber = packet
|
||||
}
|
||||
s.byIdentity[cmd.Identity] = current
|
||||
s.mu.Unlock()
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusBadGateway, current)
|
||||
writeJSON(w, http.StatusBadGateway, map[string]any{"identity": cmd.Identity, "status": "dispatch_failed", "error_code": "IOT_DISPATCH_FAILED"})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusAccepted, current)
|
||||
}
|
||||
func (s *Service) commandStatus(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
http.Error(w, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
if !s.authorized(r) {
|
||||
http.Error(w, "unauthorized", 401)
|
||||
return
|
||||
}
|
||||
identity := strings.TrimPrefix(r.URL.Path, "/v1/device-commands/")
|
||||
s.mu.RLock()
|
||||
cmd, ok := s.byIdentity[identity]
|
||||
s.mu.RUnlock()
|
||||
if !ok {
|
||||
http.Error(w, "not found", 404)
|
||||
return
|
||||
}
|
||||
writeJSON(w, 200, cmd)
|
||||
writeJSON(w, http.StatusAccepted, map[string]any{"identity": cmd.Identity, "status": status, "packet_number": packet})
|
||||
}
|
||||
func (s *Service) deviceMessage(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost || !s.authorized(r) {
|
||||
@@ -8,10 +8,10 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-client/internal/config"
|
||||
"git.apinb.com/heqiapp/platforms/backend/iot-gateway/internal/config"
|
||||
)
|
||||
|
||||
func TestDispatchPreservesIdempotency(t *testing.T) {
|
||||
func TestDispatchForwardsCommandMetadata(t *testing.T) {
|
||||
calls := 0
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
calls++
|
||||
@@ -28,7 +28,24 @@ func TestDispatchPreservesIdempotency(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommandsDoesNotDispatchDuplicateIdempotencyKey(t *testing.T) {
|
||||
func TestDeviceMessageForwardsToPlatformAPI(t *testing.T) {
|
||||
var received bool
|
||||
platform := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
received = r.URL.Path == "/heqi/internal/v1/iot/device-messages" && r.Header.Get("X-Heqi-Iot-Token") == "token"
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
}))
|
||||
defer platform.Close()
|
||||
srv := New(config.Config{HTTP: config.HTTP{InternalToken: "token"}, Upstream: config.Upstream{PlatformAPIURL: platform.URL, Token: "token"}})
|
||||
request := httptest.NewRequest(http.MethodPost, "/internal/v1/device-messages", bytes.NewBufferString(`{"type":"device_message","receivedAt":"2026-08-03T12:00:00Z"}`))
|
||||
request.Header.Set("X-Heqi-Iot-Token", "token")
|
||||
response := httptest.NewRecorder()
|
||||
srv.deviceMessage(response, request)
|
||||
if response.Code != http.StatusAccepted || !received {
|
||||
t.Fatalf("status=%d received=%v body=%s", response.Code, received, response.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommandsForwardsEveryRequestWithoutLocalState(t *testing.T) {
|
||||
calls := 0
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
calls++
|
||||
@@ -48,7 +65,7 @@ func TestCommandsDoesNotDispatchDuplicateIdempotencyKey(t *testing.T) {
|
||||
t.Fatalf("unexpected status %d: %s", response.Code, response.Body.String())
|
||||
}
|
||||
}
|
||||
if calls != 1 {
|
||||
t.Fatalf("duplicate command dispatched %d times", calls)
|
||||
if calls != 2 {
|
||||
t.Fatalf("stateless gateway forwarded %d requests, want 2", calls)
|
||||
}
|
||||
}
|
||||
@@ -94,7 +94,7 @@ func dispatchIoTOutbox(ctx context.Context) {
|
||||
}
|
||||
payload["identity"] = outbox.CommandIdentity
|
||||
body, _ := json.Marshal(payload)
|
||||
dispatch, err := http.NewRequestWithContext(ctx, http.MethodPost, config.Spec.IoTAPI.ClientBaseURL+"/v1/device-commands", bytes.NewReader(body))
|
||||
dispatch, err := http.NewRequestWithContext(ctx, http.MethodPost, config.Spec.IoTAPI.GatewayBaseURL+"/v1/device-commands", bytes.NewReader(body))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
@@ -102,7 +102,7 @@ func dispatchIoTOutbox(ctx context.Context) {
|
||||
dispatch.Header.Set("X-Heqi-Iot-Token", config.Spec.IoTAPI.Token)
|
||||
dispatchResponse, err := client.Do(dispatch)
|
||||
if err != nil {
|
||||
completeIoTOutbox(ctx, client, outbox.Identity, false, 0, "IOT_CLIENT_UNAVAILABLE")
|
||||
completeIoTOutbox(ctx, client, outbox.Identity, false, 0, "IOT_GATEWAY_UNAVAILABLE")
|
||||
continue
|
||||
}
|
||||
var result struct {
|
||||
@@ -114,7 +114,7 @@ func dispatchIoTOutbox(ctx context.Context) {
|
||||
success := (status == http.StatusAccepted || status == http.StatusOK) && decodeErr == nil
|
||||
errorCode := ""
|
||||
if !success {
|
||||
errorCode = "IOT_CLIENT_REJECTED"
|
||||
errorCode = "IOT_GATEWAY_REJECTED"
|
||||
}
|
||||
completeIoTOutbox(ctx, client, outbox.Identity, success, result.PacketNumber, errorCode)
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ PaymentAPI:
|
||||
IntervalSeconds: 60
|
||||
IoTAPI:
|
||||
PlatformBaseURL: http://127.0.0.1:12426
|
||||
ClientBaseURL: http://127.0.0.1:12429
|
||||
GatewayBaseURL: http://127.0.0.1:12429
|
||||
Token: change-me-iot-internal-token
|
||||
IntervalMilliseconds: 500
|
||||
SecretKey: change-me-to-a-random-string
|
||||
|
||||
@@ -23,7 +23,7 @@ type PaymentAPIConfig struct {
|
||||
}
|
||||
type IoTAPIConfig struct {
|
||||
PlatformBaseURL string `yaml:"PlatformBaseURL"`
|
||||
ClientBaseURL string `yaml:"ClientBaseURL"`
|
||||
GatewayBaseURL string `yaml:"GatewayBaseURL"`
|
||||
Token string `yaml:"Token"`
|
||||
IntervalMilliseconds int `yaml:"IntervalMilliseconds"`
|
||||
}
|
||||
@@ -37,7 +37,7 @@ func New(srvKey string) {
|
||||
if Spec.PaymentAPI.BaseURL == "" || Spec.PaymentAPI.Token == "" || Spec.PaymentAPI.IntervalSeconds <= 0 {
|
||||
panic("PaymentAPI configuration is required")
|
||||
}
|
||||
if Spec.IoTAPI.PlatformBaseURL == "" || Spec.IoTAPI.ClientBaseURL == "" || Spec.IoTAPI.Token == "" || Spec.IoTAPI.IntervalMilliseconds <= 0 {
|
||||
if Spec.IoTAPI.PlatformBaseURL == "" || Spec.IoTAPI.GatewayBaseURL == "" || Spec.IoTAPI.Token == "" || Spec.IoTAPI.IntervalMilliseconds <= 0 {
|
||||
panic("IoTAPI configuration is required")
|
||||
}
|
||||
conf.PrintInfo(Spec.Addr)
|
||||
|
||||
Reference in New Issue
Block a user