diff --git a/.env.example b/.env.example index 721e2a3..1a9cd0a 100644 --- a/.env.example +++ b/.env.example @@ -23,3 +23,9 @@ GIN_MODE=debug # REDIS_PORT=6379 # REDIS_PASSWORD= # REDIS_DATABASE=1 + +# S3 (local development only; production uses EC2 IAM role) +# AWS_REGION=eu-central-1 +# S3_BUCKET=luxsin-app-bucket +# AWS_ACCESS_KEY_ID=your-access-key-id +# AWS_SECRET_ACCESS_KEY=your-secret-access-key diff --git a/.env.production.example b/.env.production.example index b8f307d..ccd1964 100644 --- a/.env.production.example +++ b/.env.production.example @@ -10,3 +10,7 @@ GIN_MODE=release # root DATABASE_PASSWORD=your-production-password + +# S3 (uses EC2 IAM role; do not set access keys in production) +AWS_REGION=eu-central-1 +S3_BUCKET=luxsin-app-bucket diff --git a/cmd/server/main.go b/cmd/server/main.go index b09b157..c2e1f5b 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -18,6 +18,7 @@ import ( "github.com/luxsin/app-api/internal/repository" "github.com/luxsin/app-api/internal/router" "github.com/luxsin/app-api/internal/search" + "github.com/luxsin/app-api/internal/storage" "github.com/luxsin/app-api/internal/task" "github.com/luxsin/app-api/pkg/logger" "github.com/redis/go-redis/v9" @@ -78,7 +79,16 @@ func main() { log.Info("device persist task disabled") } - engine := router.New(log, db, searchClient, rdb, cfg.Equalize) + s3Storage, err := storage.NewS3Storage(context.Background(), cfg.S3) + if err != nil { + log.Fatal("s3 client init failed", zap.Error(err)) + } + log.Info("s3 configured", + zap.String("bucket", cfg.S3.Bucket), + zap.String("region", cfg.S3.Region), + ) + + engine := router.New(log, db, searchClient, rdb, cfg.Equalize, s3Storage) srv := &http.Server{ Addr: cfg.Addr(), diff --git a/docker-compose.yml b/docker-compose.yml index bc6a7c0..7b87d3b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,6 +15,8 @@ services: - REDIS_PASSWORD=${REDIS_PASSWORD:-eafon123!} - MEILISEARCH_API_KEY=${MEILISEARCH_API_KEY:-young9#!UJsD219921031} - ENABLE_PERSIST_TASK=false + - AWS_REGION=eu-central-1 + - S3_BUCKET=luxsin-app-bucket - MEASUREMENT_BASE_PATH=/app/measurements - TARGET_BASE_PATH=/app/targets volumes: diff --git a/go.mod b/go.mod index 2decc11..27c988e 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,10 @@ module github.com/luxsin/app-api go 1.24.0 require ( + github.com/aws/aws-sdk-go-v2 v1.41.12 + github.com/aws/aws-sdk-go-v2/config v1.32.23 + github.com/aws/aws-sdk-go-v2/credentials v1.19.22 + github.com/aws/aws-sdk-go-v2/service/s3 v1.103.2 github.com/gin-gonic/gin v1.10.0 github.com/go-sql-driver/mysql v1.10.0 github.com/meilisearch/meilisearch-go v0.36.2 @@ -13,6 +17,20 @@ require ( require ( filippo.io/edwards25519 v1.2.0 // indirect github.com/andybalholm/brotli v1.1.1 // indirect + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.28 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.29 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.21 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.28 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.28 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.1.4 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.31.2 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.5 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.43.2 // indirect + github.com/aws/smithy-go v1.27.1 // indirect github.com/bytedance/sonic v1.11.6 // indirect github.com/bytedance/sonic/loader v0.1.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect diff --git a/go.sum b/go.sum index aa25acd..8f44e70 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,42 @@ filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo= filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc= github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= +github.com/aws/aws-sdk-go-v2 v1.41.12 h1:DIKX2c31ekm9RA2D9FBj1EWXx++9AdAqRw+e78Tq2Ck= +github.com/aws/aws-sdk-go-v2 v1.41.12/go.mod h1:27+ACypSLljLAEKsCYOmrjKh83vuTRkuAe9Uv/3A4bg= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13 h1:p1BBrg/Hhp6uK7zpejeI8QFXHJeC/mynzi04Sl03k9g= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13/go.mod h1:8cIfkE9MDhkRZGpQ22aV6/lkYeYSozpz16Smrs5x4Ls= +github.com/aws/aws-sdk-go-v2/config v1.32.23 h1:PYDobtcsJXK6bQe9I8RQk6s19Bz3xa3xRU08Hy1Em3Y= +github.com/aws/aws-sdk-go-v2/config v1.32.23/go.mod h1:QID4dqUQVgEOYPKsPWd1sNWCCR2c5g7o3jeEtIXPOZU= +github.com/aws/aws-sdk-go-v2/credentials v1.19.22 h1:SHfH6wyPsEgG7fVsi5rQxWEt7tuIcN2PGhb1mTFv6tE= +github.com/aws/aws-sdk-go-v2/credentials v1.19.22/go.mod h1:54nO8lKD4aQPOntM/VTWjnR+DYzTwx0YkSMZMhAgewQ= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28 h1:b+kcDejJrXc30zU/w8Tc9klISwaO5wh+6T0sMBdDoHM= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28/go.mod h1:LnI62O9GnSv6GcuLXxOYqlq0C8EmxMcgnF6m7LdYuOY= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28 h1:Xf2j7NdVcUKomlZ4iihOP4AZ3Fzlr8h4yKpXeP+OFPg= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28/go.mod h1:O8cDo1dW63jU7ki//kRe1z+tLGcpnD1jrouitsQddDw= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.28 h1:KqIfN9kpkKkcBqBbNpNGTIrXO6ExTUvFKvXkC+YAzVo= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.28/go.mod h1:uxtQiKvLtNS4iXVsH2McVD/ls8FKN/uUhe1hGxPjrw0= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.29 h1:VkE9FuzTQVjBBrnj4+oCdxCLFIz7aqLYKUCjtvxVcOs= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.29/go.mod h1:H32Z2Qth9b+9LqjyBsCnozMQ8H2N7YBUDVXwbs0iggg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 h1:ZD2+BSw9vFsNlKYIasSNt3uDbjqqXIBcM13UJv/Lx2k= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12/go.mod h1:Ms4zlcVBbXbiP7EVLhl+lgjvA/a7YphqQ3Ih3174EmI= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.21 h1:FsZxbPiVgEHYofziwfylouMki8b1Z7mI4CMU/7bhwBA= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.21/go.mod h1:Mmm30OV+JLXYQUcbSd84THnv3P5JtjhVDujLwMqRG0U= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.28 h1:axj4mEDletwKmTm/9jR+DkIMmCfcn5vE4jBMAAN+3Vg= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.28/go.mod h1:3Aaz69M0jqfSHLKqxgolgUBFT4hpwSNc7DzC95orEi8= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.28 h1:li8rTZAAb22g4UsxbjwMdaNVWbgVcDzPqI7nDTI+mF4= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.28/go.mod h1:/brXioSGIMEdcBFoubpSdmighSVp6poP+mma/wB7iHA= +github.com/aws/aws-sdk-go-v2/service/s3 v1.103.2 h1:b4ikkRk22T4xYkEgaWc3Voe+3xbt5YbbFhNehOWyUiY= +github.com/aws/aws-sdk-go-v2/service/s3 v1.103.2/go.mod h1:Gp7eHZ0NZ8ZK5RXpoIUp/C8OeAmJqpCgdwEK1D/QOek= +github.com/aws/aws-sdk-go-v2/service/signin v1.1.4 h1:YcpVyIPLCbiypN6KSphijN5fC7DDjX114SqA7prnnxg= +github.com/aws/aws-sdk-go-v2/service/signin v1.1.4/go.mod h1:5ZICS++oFTRPfa1GsBqFDWX/8WamZ/QQOcCzIuU/zLw= +github.com/aws/aws-sdk-go-v2/service/sso v1.31.2 h1:ySNWu7TPmj5fKFIa1GYvX+Ddxd5ccruqC20aMNuyWDM= +github.com/aws/aws-sdk-go-v2/service/sso v1.31.2/go.mod h1:A+U9luAOwFeB1kseyWCITVg7/NntoPebCFR9pQ4ch9A= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.5 h1:KSzGGqfk39O+WU3OEyYbx6F7sLDQCqxlOJ+2IksfK6U= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.5/go.mod h1:ATs88lXDeQB6CZOgQ5BIl9JbYS+EsCWUSDyff6L/oVo= +github.com/aws/aws-sdk-go-v2/service/sts v1.43.2 h1:RTO7mmGyedgnNmcPh3yQizNfc6GKoV5iqfdJavuf9vw= +github.com/aws/aws-sdk-go-v2/service/sts v1.43.2/go.mod h1:fBhUZXDin9YYqhcpOMjIcpdik25rVwWyxLdPH1RZd9s= +github.com/aws/smithy-go v1.27.1 h1:4T340VFndXtADGF52gYa1POyL7s9E4Z1OeZ1hCscIw8= +github.com/aws/smithy-go v1.27.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= diff --git a/internal/config/config.go b/internal/config/config.go index 161e047..7593275 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -14,6 +14,7 @@ type Config struct { Meilisearch MeilisearchConfig Redis RedisConfig Equalize EqualizeConfig + S3 S3Config EnablePersistTask bool } @@ -44,6 +45,7 @@ func Load() (*Config, error) { } eq := loadEqualize(env) + s3cfg := loadS3(env) return &Config{ Env: env, @@ -53,6 +55,7 @@ func Load() (*Config, error) { Meilisearch: ms, Redis: rd, Equalize: eq, + S3: s3cfg, EnablePersistTask: getEnv("ENABLE_PERSIST_TASK", "false") == "true", }, nil } diff --git a/internal/config/s3.go b/internal/config/s3.go new file mode 100644 index 0000000..f226b33 --- /dev/null +++ b/internal/config/s3.go @@ -0,0 +1,36 @@ +package config + +import "os" + +type S3Config struct { + Bucket string + Region string + AccessKeyID string + SecretAccessKey string +} + +func loadS3(env string) S3Config { + if os.Getenv("AWS_ACCESS_KEY_ID") != "" || os.Getenv("AWS_SECRET_ACCESS_KEY") != "" { + return S3Config{ + Bucket: getEnv("S3_BUCKET", "luxsin-app-bucket"), + Region: getEnv("AWS_REGION", "eu-central-1"), + AccessKeyID: os.Getenv("AWS_ACCESS_KEY_ID"), + SecretAccessKey: os.Getenv("AWS_SECRET_ACCESS_KEY"), + } + } + + switch env { + case "production": + return S3Config{ + Bucket: getEnv("S3_BUCKET", "luxsin-app-bucket"), + Region: getEnv("AWS_REGION", "eu-central-1"), + } + default: + return S3Config{ + Bucket: getEnv("S3_BUCKET", "luxsin-app-bucket"), + Region: getEnv("AWS_REGION", "eu-central-1"), + AccessKeyID: "AKIAVMFK45I6P3TNI3VJ", + SecretAccessKey: "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", + } + } +} diff --git a/internal/handler/curve.go b/internal/handler/curve.go index c478e75..507dbf7 100644 --- a/internal/handler/curve.go +++ b/internal/handler/curve.go @@ -30,10 +30,53 @@ type CurveHandler struct { log *zap.Logger } +const defaultModelCurveTarget = "Harman over-ear 2018" + func NewCurveHandler(repo *repository.CurveRepository, curveCache *cache.CurveCache, cfg config.EqualizeConfig, log *zap.Logger) *CurveHandler { return &CurveHandler{repo: repo, cache: curveCache, cfg: cfg, log: log} } +// ModelCurve 获取机型默认曲线（固定 target: Harman over-ear 2018） +// GET /audio/modelCurve?brand=xxx&name=xxx&base64Resp=true +func (h *CurveHandler) ModelCurve(c *gin.Context) { + brand := strings.TrimSpace(queryParam(c, "brand")) + name := strings.TrimSpace(queryParam(c, "name")) + base64Resp := encode.ParseBase64Param(c) + + if brand == "" || name == "" { + h.writeCurveResponse(c, base64Resp, gin.H{ + "code": 400, + "msg": "参数校验失败", + }) + return + } + + ctx := c.Request.Context() + + result, err := h.getCurvePoint(ctx, brand, name, defaultModelCurveTarget) + if err != nil { + h.log.Error("get model curve failed", zap.Error(err)) + response.InternalError(c, "获取曲线数据失败") + return + } + + if result == "" { + h.writeCurveResponse(c, base64Resp, gin.H{ + "code": 0, + "msg": "无曲线数据", + }) + return + } + + var resp map[string]any + if err := json.Unmarshal([]byte(result), &resp); err != nil { + response.InternalError(c, "解析曲线数据失败") + return + } + + h.writeCurveResponse(c, base64Resp, resp) +} + // GetCurve 获取目标曲线 // GET /audio/getCurve?brand=xxx&name=xxx&target=xxx&base64Resp=true func (h *CurveHandler) GetCurve(c *gin.Context) { @@ -57,7 +100,7 @@ func (h *CurveHandler) GetCurve(c *gin.Context) { } if result == "" { - response.Fail(c, http.StatusOK, 40004, "无曲线数据") + response.Fail(c, http.StatusOK, 0, "无曲线数据") return } @@ -90,6 +133,20 @@ func (h *CurveHandler) GetCurve(c *gin.Context) { c.JSON(http.StatusOK, resultData) } +func (h *CurveHandler) writeCurveResponse(c *gin.Context, base64Resp bool, data any) { + if base64Resp { + encoded, err := encode.EncodeJSON(data) + if err != nil { + h.log.Error("encode response failed", zap.Error(err)) + response.InternalError(c, "编码响应失败") + return + } + c.String(http.StatusOK, encoded) + return + } + c.JSON(http.StatusOK, data) +} + // getCurvePoint 获取曲线数据：先查缓存，缓存不存在则请求 EQ 接口并缓存结果 func (h *CurveHandler) getCurvePoint(ctx context.Context, brand, name, target string) (string, error) { data, acquired, err := h.cache.GetWithLock(ctx, brand, name, target) diff --git a/internal/handler/model_csv.go b/internal/handler/model_csv.go new file mode 100644 index 0000000..ffd57dc --- /dev/null +++ b/internal/handler/model_csv.go @@ -0,0 +1,161 @@ +package handler + +import ( + "bytes" + "encoding/csv" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "unicode" + + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/gin-gonic/gin" + "github.com/luxsin/app-api/internal/storage" + "github.com/luxsin/app-api/pkg/encode" + "go.uber.org/zap" +) + +type ModelCSVHandler struct { + s3 *storage.S3Storage + log *zap.Logger +} + +func NewModelCSVHandler(s3 *storage.S3Storage, log *zap.Logger) *ModelCSVHandler { + return &ModelCSVHandler{s3: s3, log: log} +} + +// GetModelCSV 从 S3 读取耳机 CSV 频响数据 +// GET /audio/getModelCSV?brand=Abyss&model=Dinan DZ&form=over-ear&base64Resp=true +func (h *ModelCSVHandler) GetModelCSV(c *gin.Context) { + brand := strings.TrimSpace(queryParam(c, "brand")) + model := strings.TrimSpace(queryParam(c, "model")) + form := strings.TrimSpace(queryParam(c, "form")) + base64Resp := encode.ParseBase64Param(c) + + if brand == "" || model == "" || form == "" { + h.writeModelCSVResponse(c, base64Resp, gin.H{ + "code": 400, + "msg": "参数校验失败", + }) + return + } + + key := modelCSVKey(brand, model, form) + ctx := c.Request.Context() + + data, err := h.s3.GetObject(ctx, key) + if err != nil { + var noSuchKey *types.NoSuchKey + if errors.As(err, &noSuchKey) { + h.log.Info("model csv not found in s3", zap.String("key", key)) + h.writeModelCSVResponse(c, base64Resp, gin.H{ + "code": 0, + "msg": "无曲线数据", + }) + return + } + h.log.Error("get model csv from s3 failed", + zap.String("key", key), + zap.Error(err), + ) + h.writeModelCSVResponse(c, base64Resp, gin.H{ + "code": 500, + "msg": "系统错误", + }) + return + } + + parsed, err := parseCSVData(data) + if err != nil || parsed == nil { + h.log.Error("parse model csv failed", zap.String("key", key), zap.Error(err)) + h.writeModelCSVResponse(c, base64Resp, gin.H{ + "code": 0, + "msg": "无曲线数据", + }) + return + } + + h.writeModelCSVResponse(c, base64Resp, gin.H{ + "code": 200, + "msg": "操作成功", + "frequency": parsed["frequency"], + "raw": parsed["raw"], + }) +} + +func modelCSVKey(brand, model, form string) string { + filename := brand + " " + model + ".csv" + return fmt.Sprintf("autoeq/measurements/Eafonyoung/data/%s/%s/%s", + form, brandPrefix(brand), filename) +} + +func brandPrefix(brand string) string { + runes := []rune(brand) + if len(runes) == 0 { + return "" + } + first := runes[0] + if unicode.IsLetter(first) && unicode.IsLower(first) { + return strings.ToUpper(string(first)) + } + return string(first) +} + +func parseCSVData(data []byte) (map[string]any, error) { + reader := csv.NewReader(bytes.NewReader(data)) + if _, err := reader.Read(); err != nil { + return nil, fmt.Errorf("read csv header: %w", err) + } + + frequency := make([]float64, 0) + raw := make([]float64, 0) + + for { + record, err := reader.Read() + if err == io.EOF { + break + } + if err != nil { + return nil, fmt.Errorf("read csv row: %w", err) + } + if len(record) < 2 { + continue + } + freq, err1 := strconv.ParseFloat(record[0], 64) + val, err2 := strconv.ParseFloat(record[1], 64) + if err1 != nil || err2 != nil { + continue + } + frequency = append(frequency, freq) + raw = append(raw, val) + } + + if len(frequency) == 0 { + return nil, nil + } + + return map[string]any{ + "frequency": frequency, + "raw": raw, + }, nil +} + +func (h *ModelCSVHandler) writeModelCSVResponse(c *gin.Context, base64Resp bool, data any) { + if base64Resp { + encoded, err := encode.EncodeJSON(data) + if err != nil { + h.log.Error("encode response failed", zap.Error(err)) + c.JSON(http.StatusOK, gin.H{ + "code": 500, + "msg": "系统错误", + }) + return + } + c.String(http.StatusOK, encoded) + return + } + c.JSON(http.StatusOK, data) +} diff --git a/internal/router/router.go b/internal/router/router.go index 7628bad..6547d14 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -10,11 +10,12 @@ import ( "github.com/luxsin/app-api/internal/middleware" "github.com/luxsin/app-api/internal/repository" "github.com/luxsin/app-api/internal/search" + "github.com/luxsin/app-api/internal/storage" "github.com/redis/go-redis/v9" "go.uber.org/zap" ) -func New(log *zap.Logger, db *sql.DB, searchClient *search.Client, rdb *redis.Client, eqCfg config.EqualizeConfig) *gin.Engine { +func New(log *zap.Logger, db *sql.DB, searchClient *search.Client, rdb *redis.Client, eqCfg config.EqualizeConfig, s3 *storage.S3Storage) *gin.Engine { r := gin.New() r.Use(gin.Recovery()) r.Use(middleware.RequestID()) @@ -40,6 +41,7 @@ func New(log *zap.Logger, db *sql.DB, searchClient *search.Client, rdb *redis.Cl device := handler.NewDeviceHandler(rdb, log) ota := handler.NewOTAHandler(otaRepo, log) curve := handler.NewCurveHandler(curveRepo, curveCache, eqCfg, log) + modelCSV := handler.NewModelCSVHandler(s3, log) v1 := r.Group("/api/v1") { @@ -54,6 +56,8 @@ func New(log *zap.Logger, db *sql.DB, searchClient *search.Client, rdb *redis.Cl audio.GET("/reportDevInfo", device.ReportDevInfo) audio.GET("/ota", ota.GetOTA) audio.GET("/getCurve", curve.GetCurve) + audio.GET("/modelCurve", curve.ModelCurve) + audio.GET("/getModelCSV", modelCSV.GetModelCSV) } return r diff --git a/internal/storage/s3.go b/internal/storage/s3.go new file mode 100644 index 0000000..4099780 --- /dev/null +++ b/internal/storage/s3.go @@ -0,0 +1,56 @@ +package storage + +import ( + "context" + "fmt" + "io" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/luxsin/app-api/internal/config" +) + +type S3Storage struct { + client *s3.Client + bucket string +} + +func NewS3Storage(ctx context.Context, cfg config.S3Config) (*S3Storage, error) { + var opts []func(*awsconfig.LoadOptions) error + opts = append(opts, awsconfig.WithRegion(cfg.Region)) + + if cfg.AccessKeyID != "" && cfg.SecretAccessKey != "" { + opts = append(opts, awsconfig.WithCredentialsProvider( + credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, ""), + )) + } + + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, opts...) + if err != nil { + return nil, fmt.Errorf("load aws config: %w", err) + } + + return &S3Storage{ + client: s3.NewFromConfig(awsCfg), + bucket: cfg.Bucket, + }, nil +} + +func (s *S3Storage) GetObject(ctx context.Context, key string) ([]byte, error) { + out, err := s.client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(key), + }) + if err != nil { + return nil, err + } + defer out.Body.Close() + + data, err := io.ReadAll(out.Body) + if err != nil { + return nil, fmt.Errorf("read s3 object body: %w", err) + } + return data, nil +} diff --git a/internal/task/device_persist.go b/internal/task/device_persist.go index 31ad895..bad0030 100644 --- a/internal/task/device_persist.go +++ b/internal/task/device_persist.go @@ -80,14 +80,32 @@ func (t *DevicePersistTask) Persist() { } } - // 批量删除成功处理的记录 - if len(succeeded) > 0 { + total := len(devices) + persisted := len(succeeded) + failed := total - persisted + + // 批量删除成功刷入的记录，失败的保留在 Redis 等待下次重试 + if persisted > 0 { if err := t.rdb.HDel(ctx, "devices", succeeded...).Err(); err != nil { - t.log.Error("redis HDel failed", zap.Int("count", len(succeeded)), zap.Error(err)) + t.log.Error("redis HDel failed after persist", + zap.Int("persisted", persisted), + zap.Error(err), + ) } else { - t.log.Info("devices persisted and removed from redis", zap.Int("count", len(succeeded))) + t.log.Info("devices persisted to database and removed from redis", + zap.Int("persisted", persisted), + zap.Int("total", total), + zap.Int("failed", failed), + ) } } + + if failed > 0 { + t.log.Warn("some devices failed to persist, kept in redis for retry", + zap.Int("failed", failed), + zap.Int("total", total), + ) + } } // persistDevice 处理 user_device 表写入 diff --git a/scripts/upload.sh b/scripts/upload.sh new file mode 100755 index 0000000..7885f24 --- /dev/null +++ b/scripts/upload.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +set -euo pipefail + +REMOTE="ubuntu@ec2-18-184-205-87.eu-central-1.compute.amazonaws.com:/data/project/app-api/" +KEY="$HOME/.ssh/aws_eafon.pem" + +if [ $# -eq 0 ]; then + PATHS=( + ./cmd/ + ./internal/ + ./pkg + ./go.mod + ./go.sum + ./Dockerfile + ./docker-compose.yml + ) +else + PATHS=("$@") +fi + +scp -i "$KEY" -r "${PATHS[@]}" "$REMOTE" diff --git a/upload.sh b/upload.sh deleted file mode 100755 index e03c6e1..0000000 --- a/upload.sh +++ /dev/null @@ -1 +0,0 @@ -scp -i ~/.ssh/aws_eafon.pem -r ./cmd/ ./internal/ ./pkg ./go.mod ./go.sum ./Dockerfile ./docker-compose.yml ubuntu@ec2-18-184-205-87.eu-central-1.compute.amazonaws.com:/data/project/app-api/