From 7b942da97574b60bf6a316b00609fb18896e60f7 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 23 Aug 2026 23:20:33 +0800 Subject: [PATCH] feat: add aliyun oss media handler --- go.mod | 1 + go.sum | 2 + server/main.go | 1 + server/media/oss/oss.go | 262 ++++++++++++++++++++++++++++ server/media/oss/oss_tenant_test.go | 34 ++++ 5 files changed, 300 insertions(+) create mode 100644 server/media/oss/oss.go create mode 100644 server/media/oss/oss_tenant_test.go diff --git a/go.mod b/go.mod index a5a33a3..2055fb5 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.0 require ( firebase.google.com/go v3.13.0+incompatible + github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible github.com/aws/aws-sdk-go-v2 v1.42.1 github.com/aws/aws-sdk-go-v2/config v1.32.27 github.com/aws/aws-sdk-go-v2/credentials v1.19.26 diff --git a/go.sum b/go.sum index 5459efb..13f69c7 100644 --- a/go.sum +++ b/go.sum @@ -35,6 +35,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.57.0/go.mod h1:dzcEjy1WJ0Q4u9twNR3LcLhNoYMRCrMCMafpxa0TjPQ= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0 h1:RoO5+d7uCmDqovLrHCr2/BuViUXvdcrNxyNM1pN9dDQ= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0/go.mod h1:YqwkQPrWSC7+byyc1VlKbWLBF5JsW5IoL6xUkemYSXk= +github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible h1:8psS8a+wKfiLt1iVDX79F7Y6wUM49Lcha2FMXt4UM8g= +github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8= github.com/aws/aws-sdk-go-v2 v1.42.1 h1:9eOTgu1z/dVtYpNZ3/8/XbbaX0x/BqE3HUzAzs6K0ek= github.com/aws/aws-sdk-go-v2 v1.42.1/go.mod h1:5pKeft2eJj+gElQ38Jqg4ibCqh+/AK33/0X3hip7IjM= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14 h1:3IZY0XAJquT3aHzbkHfPzy4ACPcEjVG0x87KOwtpqGY= diff --git a/server/main.go b/server/main.go index 85bff4b..b579157 100644 --- a/server/main.go +++ b/server/main.go @@ -57,6 +57,7 @@ import ( // File upload handlers _ "github.com/tinode/chat/server/media/fs" + _ "github.com/tinode/chat/server/media/oss" _ "github.com/tinode/chat/server/media/s3" ) diff --git a/server/media/oss/oss.go b/server/media/oss/oss.go new file mode 100644 index 0000000..bb326b6 --- /dev/null +++ b/server/media/oss/oss.go @@ -0,0 +1,262 @@ +// Package oss implements media interface by storing media objects in Alibaba Cloud OSS. +package oss + +import ( + "encoding/base32" + "encoding/json" + "errors" + "hash/fnv" + "io" + "mime" + "net/http" + "net/url" + "path" + "strconv" + "strings" + "sync/atomic" + "time" + + aliyunoss "github.com/aliyun/aliyun-oss-go-sdk/oss" + + "github.com/tinode/chat/server/logs" + "github.com/tinode/chat/server/media" + "github.com/tinode/chat/server/store" + "github.com/tinode/chat/server/store/types" +) + +const ( + defaultServeURL = "/v0/file/s/" + defaultCacheControl = "no-cache, must-revalidate" + + handlerName = "oss" + // Presign GET URLs for this number of seconds. + defaultPresignDuration = 120 +) + +type ossConfig struct { + AccessKeyID string `json:"access_key_id"` + AccessKeySecret string `json:"access_key_secret"` + Endpoint string `json:"endpoint"` + BucketName string `json:"bucket"` + CorsOrigins []string `json:"cors_origins"` + ServeURL string `json:"serve_url"` + PresignTTL int `json:"presign_ttl"` + CacheControl string `json:"cache_control"` + VerifyBucket bool `json:"verify_bucket"` +} + +type ossHandler struct { + client *aliyunoss.Client + bucket *aliyunoss.Bucket + conf ossConfig + corsOrigins []media.AllowedOrigin +} + +type readerCounter struct { + count int64 + reader io.Reader +} + +func (rc *readerCounter) Read(buf []byte) (int, error) { + n, err := rc.reader.Read(buf) + atomic.AddInt64(&rc.count, int64(n)) + return n, err +} + +// Init initializes the media handler. +func (oh *ossHandler) Init(jsconf string) error { + var err error + if err = json.Unmarshal([]byte(jsconf), &oh.conf); err != nil { + return errors.New("failed to parse config: " + err.Error()) + } + + if oh.conf.AccessKeyID == "" { + return errors.New("missing Access Key ID") + } + if oh.conf.AccessKeySecret == "" { + return errors.New("missing Access Key Secret") + } + if oh.conf.Endpoint == "" { + return errors.New("missing Endpoint") + } + if oh.conf.BucketName == "" { + return errors.New("missing Bucket") + } + if oh.conf.PresignTTL <= 0 { + oh.conf.PresignTTL = defaultPresignDuration + } + if oh.conf.CacheControl == "" { + oh.conf.CacheControl = defaultCacheControl + } + if oh.conf.ServeURL == "" { + oh.conf.ServeURL = defaultServeURL + } + + oh.corsOrigins, err = media.ParseCORSAllow(oh.conf.CorsOrigins) + if err != nil { + return errors.New("failed to parse CORS allowed origins: " + err.Error()) + } + + oh.client, err = aliyunoss.New(oh.conf.Endpoint, oh.conf.AccessKeyID, oh.conf.AccessKeySecret) + if err != nil { + return err + } + if oh.conf.VerifyBucket { + exists, err := oh.client.IsBucketExist(oh.conf.BucketName) + if err != nil { + return err + } + if !exists { + return errors.New("bucket does not exist") + } + } + oh.bucket, err = oh.client.Bucket(oh.conf.BucketName) + return err +} + +// Headers adds CORS headers and redirects GET and HEAD requests to OSS. +func (oh *ossHandler) Headers(tenantID types.TenantID, method string, url *url.URL, headers http.Header, serve bool) (http.Header, int, error) { + headers, status := media.CORSHandler(method, headers, oh.corsOrigins, serve) + if status != 0 || method == http.MethodPost || method == http.MethodPut { + return headers, status, nil + } + + fid := oh.GetIdFromUrl(url.String()) + if fid.IsZero() { + return nil, 0, types.ErrNotFound + } + + fdef, err := oh.getFileRecord(tenantID, fid) + if err != nil { + return nil, 0, err + } + + if fdef.ETag != "" && strings.Trim(headers.Get("If-None-Match"), "\"") == fdef.ETag { + return http.Header{ + "ETag": {`"` + fdef.ETag + `"`}, + "Cache-Control": {oh.conf.CacheControl}, + }, http.StatusNotModified, nil + } + + var redirURL string + switch method { + case http.MethodGet: + options := []aliyunoss.Option{ + aliyunoss.ResponseCacheControl(oh.conf.CacheControl), + aliyunoss.ResponseContentType(fdef.MimeType), + } + if isAttachment, _ := strconv.ParseBool(url.Query().Get("asatt")); isAttachment { + options = append(options, aliyunoss.ResponseContentDisposition("attachment")) + } + redirURL, err = oh.bucket.SignURL(fdef.Location, aliyunoss.HTTPGet, int64(oh.conf.PresignTTL), options...) + case http.MethodHead: + redirURL, err = oh.bucket.SignURL(fdef.Location, aliyunoss.HTTPHead, int64(oh.conf.PresignTTL)) + } + if err != nil { + return nil, 0, err + } + + if redirURL != "" { + return http.Header{ + "Location": {redirURL}, + "ETag": {`"` + fdef.ETag + `"`}, + "Content-Type": {"application/json; charset=utf-8"}, + "Cache-Control": {oh.conf.CacheControl}, + }, http.StatusPermanentRedirect, nil + } + return nil, 0, nil +} + +// Upload processes request for a file upload. +func (oh *ossHandler) Upload(fdef *types.FileDef, file io.Reader) (string, int64, error) { + key, err := tenantObjectKey(fdef.TenantID, fdef.Uid().String32(), fdef.CreatedAt) + if err != nil { + return "", 0, err + } + + if err = store.Files.StartUpload(fdef.TenantID, fdef); err != nil { + logs.Warn.Println("failed to create file record", fdef.Id, err) + return "", 0, err + } + + rc := readerCounter{reader: file} + options := []aliyunoss.Option{aliyunoss.CacheControl(oh.conf.CacheControl)} + if fdef.MimeType != "" { + options = append(options, aliyunoss.ContentType(fdef.MimeType)) + } + if err = oh.bucket.PutObject(key, &rc, options...); err != nil { + return "", 0, err + } + + fname := fdef.Id + ext, _ := mime.ExtensionsByType(fdef.MimeType) + if len(ext) > 0 { + fname += ext[0] + } + + fdef.Location = key + fdef.ETag = etagFromLocation(key) + if meta, err := oh.bucket.GetObjectDetailedMeta(key); err == nil { + if etag := strings.Trim(meta.Get("ETag"), "\""); etag != "" { + fdef.ETag = etag + } + } + return oh.conf.ServeURL + fname, rc.count, nil +} + +// Download is unsupported because OSS objects are served through signed redirects. +func (oh *ossHandler) Download(tenantID types.TenantID, url string) (*types.FileDef, media.ReadSeekCloser, error) { + return nil, nil, types.ErrUnsupported +} + +// Delete deletes files from OSS by provided object keys. +func (oh *ossHandler) Delete(locations []string) error { + for i := 0; i < len(locations); i += 1000 { + end := i + 1000 + if end > len(locations) { + end = len(locations) + } + if _, err := oh.bucket.DeleteObjects(locations[i:end], aliyunoss.DeleteObjectsQuiet(true)); err != nil { + return err + } + } + return nil +} + +// GetIdFromUrl converts an attachment URL to a file UID. +func (oh *ossHandler) GetIdFromUrl(url string) types.Uid { + return media.GetIdFromUrl(url, oh.conf.ServeURL) +} + +func (oh *ossHandler) getFileRecord(tenantID types.TenantID, fid types.Uid) (*types.FileDef, error) { + fd, err := store.Files.Get(tenantID, fid.String()) + if err != nil { + return nil, err + } + if fd == nil { + return nil, types.ErrNotFound + } + return fd, nil +} + +func tenantObjectKey(tenantID types.TenantID, fileID string, createdAt time.Time) (string, error) { + if !tenantID.IsValid() || fileID == "" { + return "", types.ErrMalformed + } + if createdAt.IsZero() { + createdAt = time.Now().UTC() + } + return path.Join("tenants", strconv.FormatInt(int64(tenantID), 10), + createdAt.Format("2006"), createdAt.Format("01"), fileID), nil +} + +func etagFromLocation(location string) string { + hasher := fnv.New128() + hasher.Write([]byte(location)) + return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding). + EncodeToString(hasher.Sum(make([]byte, 0, hasher.Size())))) +} + +func init() { + store.RegisterMediaHandler(handlerName, &ossHandler{}) +} diff --git a/server/media/oss/oss_tenant_test.go b/server/media/oss/oss_tenant_test.go new file mode 100644 index 0000000..2134477 --- /dev/null +++ b/server/media/oss/oss_tenant_test.go @@ -0,0 +1,34 @@ +package oss + +import ( + "testing" + "time" + + "github.com/tinode/chat/server/store/types" +) + +func TestTenantObjectKey(t *testing.T) { + got, err := tenantObjectKey(types.TenantID(42), "abc123", time.Date(2026, 8, 11, 9, 0, 0, 0, time.UTC)) + if err != nil { + t.Fatalf("tenantObjectKey failed: %v", err) + } + if want := "tenants/42/2026/08/abc123"; got != want { + t.Fatalf("tenantObjectKey = %q, want %q", got, want) + } +} + +func TestTenantObjectKeyRejectsZeroTenant(t *testing.T) { + if _, err := tenantObjectKey(types.ZeroTenantID, "abc123", time.Time{}); err != types.ErrMalformed { + t.Fatalf("tenantObjectKey err = %v, want malformed", err) + } +} + +func TestGetIdFromUrl(t *testing.T) { + handler := ossHandler{conf: ossConfig{ServeURL: "/v0/file/s/"}} + + fid := types.Uid(123456) + got := handler.GetIdFromUrl("/v0/file/s/" + fid.String() + ".jpg") + if got != fid { + t.Fatalf("GetIdFromUrl = %v, want %v", got, fid) + } +}