feat: add aliyun oss media handler

2026-08-23 23:20:33 +08:00
parent a578acebf7
commit 7b942da975
5 changed files with 300 additions and 0 deletions
@@ -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"
 )

@@ -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{})
+}
@@ -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)
+	}
+}