package app import ( "bytes" "context" "database/sql" "encoding/json" "fmt" "io" "log" "net/http" "net/url" "strings" "sync" "time" "lingniu/vehicle-data-platform/apps/api/internal/config" "lingniu/vehicle-data-platform/apps/api/internal/httpx" "lingniu/vehicle-data-platform/apps/api/internal/platform" "lingniu/vehicle-data-platform/apps/api/internal/static" ) func NewServer(cfg config.Config) http.Handler { var store platform.Store = platform.NewMockStore() if cfg.MySQLDSN != "" { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() db, err := platform.OpenSQL(ctx, "mysql", cfg.MySQLDSN) if err != nil { log.Printf("production mysql store disabled: %v", err) } else { var tdengine *sql.DB if cfg.TDengineDSN != "" { tdengine, err = platform.OpenSQL(ctx, cfg.TDengineDriver, cfg.TDengineDSN) if err != nil { log.Printf("production tdengine store disabled: %v", err) } else { log.Printf("production tdengine store enabled") } } productionStore := platform.NewProductionStore(db, tdengine, cfg.TDengineDatabase) if cfg.RedisAddr != "" { productionStore.WithRedisOnlineKeyCounter(platform.NewRedisOnlineKeyCounter(cfg.RedisAddr, cfg.RedisUsername, cfg.RedisPassword, cfg.RedisDB)) log.Printf("production redis health probe enabled") } if cfg.CapacityCheckBin != "" { productionStore.WithCapacityChecker(platform.NewCapacityCheckCommand(cfg.CapacityCheckBin)) log.Printf("production capacity-check probe enabled") } store = productionStore log.Printf("production mysql store enabled") } } api := platform.NewHandler(platform.NewServiceWithRuntime(store, platform.RuntimeInfo{ RequestTimeoutMs: int(cfg.RequestTimeout / time.Millisecond), AMapWebJSConfigured: strings.TrimSpace(cfg.AMapWebJSKey) != "", AMapSecurityProxyEnabled: strings.TrimSpace(cfg.AMapSecurityCode) != "" && strings.TrimSpace(cfg.AMapServiceHost) != "", AMapSecurityCodeExposed: exposedAMapSecurityCode(cfg) != "", AMapSecurityServiceHost: strings.TrimSpace(cfg.AMapServiceHost), PlatformRelease: strings.TrimSpace(cfg.PlatformRelease), })) handler := static.Handler(cfg.StaticDir, api) handler = withAppConfig(handler, cfg) handler = withAMapSecurityProxy(handler, cfg, defaultAMapProxyUpstreams(), http.DefaultClient) return withRequestTimeout(handler, cfg.RequestTimeout) } func withAppConfig(next http.Handler, cfg config.Config) http.Handler { type appConfig struct { AMapWebJSKey string `json:"amapWebJsKey,omitempty"` AMapSecurityCode string `json:"amapSecurityJsCode,omitempty"` AMapServiceHost string `json:"amapSecurityServiceHost,omitempty"` } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/app-config.js" { next.ServeHTTP(w, r) return } w.Header().Set("Content-Type", "application/javascript; charset=utf-8") w.Header().Set("Cache-Control", "no-store") body, err := json.Marshal(appConfig{ AMapWebJSKey: cfg.AMapWebJSKey, AMapSecurityCode: exposedAMapSecurityCode(cfg), AMapServiceHost: cfg.AMapServiceHost, }) if err != nil { http.Error(w, "failed to render app config", http.StatusInternalServerError) return } _, _ = fmt.Fprintf(w, "window.__LINGNIU_APP_CONFIG__=%s;\n", body) }) } func exposedAMapSecurityCode(cfg config.Config) string { if cfg.AMapServiceHost != "" { return "" } return cfg.AMapSecurityCode } type amapProxyUpstreams struct { RestAPI string WebAPI string FMap string } func defaultAMapProxyUpstreams() amapProxyUpstreams { return amapProxyUpstreams{ RestAPI: "https://restapi.amap.com", WebAPI: "https://webapi.amap.com", FMap: "https://fmap01.amap.com", } } func (upstreams amapProxyUpstreams) normalized() amapProxyUpstreams { defaults := defaultAMapProxyUpstreams() if upstreams.RestAPI == "" { upstreams.RestAPI = defaults.RestAPI } if upstreams.WebAPI == "" { upstreams.WebAPI = defaults.WebAPI } if upstreams.FMap == "" { upstreams.FMap = defaults.FMap } upstreams.RestAPI = strings.TrimRight(upstreams.RestAPI, "/") upstreams.WebAPI = strings.TrimRight(upstreams.WebAPI, "/") upstreams.FMap = strings.TrimRight(upstreams.FMap, "/") return upstreams } func withAMapSecurityProxy(next http.Handler, cfg config.Config, upstreams amapProxyUpstreams, client *http.Client) http.Handler { serviceHost := strings.TrimRight(cfg.AMapServiceHost, "/") securityCode := strings.TrimSpace(cfg.AMapSecurityCode) if serviceHost == "" || securityCode == "" { return next } if client == nil { client = http.DefaultClient } upstreams = upstreams.normalized() return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != serviceHost && !strings.HasPrefix(r.URL.Path, serviceHost+"/") { next.ServeHTTP(w, r) return } target, err := amapProxyURL(upstreams, serviceHost, securityCode, r) if err != nil { httpx.WriteError(w, http.StatusBadGateway, "AMAP_PROXY_URL_INVALID", "高德地图代理地址无效", err.Error(), requestTraceID(r)) return } req, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body) if err != nil { httpx.WriteError(w, http.StatusBadGateway, "AMAP_PROXY_REQUEST_INVALID", "高德地图代理请求无效", err.Error(), requestTraceID(r)) return } req.Header = r.Header.Clone() req.Host = "" resp, err := client.Do(req) if err != nil { httpx.WriteError(w, http.StatusBadGateway, "AMAP_PROXY_FAILED", "高德地图代理请求失败", err.Error(), requestTraceID(r)) return } defer resp.Body.Close() for key, values := range resp.Header { for _, value := range values { w.Header().Add(key, value) } } w.WriteHeader(resp.StatusCode) _, _ = io.Copy(w, resp.Body) }) } func amapProxyURL(upstreams amapProxyUpstreams, serviceHost string, securityCode string, r *http.Request) (string, error) { targetPath := strings.TrimPrefix(r.URL.Path, serviceHost) if targetPath == "" { targetPath = "/" } upstreamBase := upstreams.RestAPI if strings.HasPrefix(targetPath, "/v4/map/styles") { upstreamBase = upstreams.WebAPI } else if strings.HasPrefix(targetPath, "/v3/vectormap") { upstreamBase = upstreams.FMap } base, err := url.Parse(upstreamBase) if err != nil { return "", err } base.Path = strings.TrimRight(base.Path, "/") + targetPath query := r.URL.Query() query.Set("jscode", securityCode) base.RawQuery = query.Encode() return base.String(), nil } func withRequestTimeout(next http.Handler, timeout time.Duration) http.Handler { if timeout <= 0 { return next } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithTimeout(r.Context(), timeout) defer cancel() buffered := newBufferedResponseWriter() result := make(chan any, 1) go func() { defer func() { result <- recover() }() next.ServeHTTP(buffered, r.WithContext(ctx)) }() select { case panicValue := <-result: if panicValue != nil { panic(panicValue) } buffered.writeTo(w) case <-ctx.Done(): httpx.WriteError(w, http.StatusServiceUnavailable, "REQUEST_TIMEOUT", "请求处理超时", "", requestTraceID(r)) } }) } func requestTraceID(r *http.Request) string { if value := r.Header.Get("X-Trace-Id"); value != "" { return value } return "trace-" + time.Now().Format("20060102150405.000000") } type bufferedResponseWriter struct { mu sync.Mutex header http.Header status int body bytes.Buffer } func newBufferedResponseWriter() *bufferedResponseWriter { return &bufferedResponseWriter{header: http.Header{}, status: http.StatusOK} } func (w *bufferedResponseWriter) Header() http.Header { return w.header } func (w *bufferedResponseWriter) WriteHeader(statusCode int) { w.mu.Lock() defer w.mu.Unlock() w.status = statusCode } func (w *bufferedResponseWriter) Write(body []byte) (int, error) { w.mu.Lock() defer w.mu.Unlock() return w.body.Write(body) } func (w *bufferedResponseWriter) writeTo(target http.ResponseWriter) { w.mu.Lock() defer w.mu.Unlock() for key, values := range w.header { for _, value := range values { target.Header().Add(key, value) } } target.WriteHeader(w.status) _, _ = target.Write(w.body.Bytes()) }