feat(storage): serve media through backend and add local FS backend
- introduce storage.Backend with S3 and local file-system implementations - deliver all blobs via signed GET/HEAD /api/objects/<key> (Range, Content-Type) - stream S3 objects through the backend; remove presigned URL path - share shared HMAC Signer between backends - remove S3_PUBLIC_URL; rename S3_PRESIGN_TTL to OBJECT_URL_TTL - make storage label in health dynamic; nginx proxy_buffering off for objects
This commit is contained in:
@@ -0,0 +1,211 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"image"
|
||||
"image/color"
|
||||
"image/png"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"shared-album/internal/albums"
|
||||
"shared-album/internal/auth"
|
||||
"shared-album/internal/config"
|
||||
"shared-album/internal/members"
|
||||
"shared-album/internal/models"
|
||||
"shared-album/internal/photos"
|
||||
"shared-album/internal/sharelinks"
|
||||
"shared-album/internal/sse"
|
||||
"shared-album/internal/storage"
|
||||
)
|
||||
|
||||
func integrationPool(t *testing.T) *pgxpool.Pool {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_DATABASE_URL not set")
|
||||
}
|
||||
pool, err := pgxpool.New(context.Background(), dsn)
|
||||
if err != nil {
|
||||
t.Skipf("cannot connect: %v", err)
|
||||
}
|
||||
if err := pool.Ping(context.Background()); err != nil {
|
||||
t.Skipf("cannot ping: %v", err)
|
||||
}
|
||||
t.Cleanup(pool.Close)
|
||||
return pool
|
||||
}
|
||||
|
||||
func makeTestPNG(t *testing.T) []byte {
|
||||
t.Helper()
|
||||
img := image.NewRGBA(image.Rect(0, 0, 4, 4))
|
||||
for x := 0; x < 4; x++ {
|
||||
for y := 0; y < 4; y++ {
|
||||
img.Set(x, y, color.RGBA{R: uint8(x * 40), G: uint8(y * 40), B: 128, A: 255})
|
||||
}
|
||||
}
|
||||
var buf bytes.Buffer
|
||||
if err := png.Encode(&buf, img); err != nil {
|
||||
t.Fatalf("encode png: %v", err)
|
||||
}
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
// runShareFlow uploads a photo and validates the read-only share flow returns a
|
||||
// usable photo URL. verify inspects/serves the URL according to the backend.
|
||||
func runShareFlow(t *testing.T, backend storage.Backend, verify func(t *testing.T, router *gin.Engine, raw string)) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
pool := integrationPool(t)
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
hub := sse.NewHub(1)
|
||||
srv := New(&config.Config{}, pool, backend, hub)
|
||||
router := srv.Router()
|
||||
|
||||
authStore := auth.NewStore(pool)
|
||||
owner := &models.User{ID: auth.NewID(), Email: auth.NewID()[:8] + "@test.local", PasswordHash: "x", Name: "Owner"}
|
||||
if err := authStore.CreateUser(ctx, owner); err != nil {
|
||||
t.Fatalf("create user: %v", err)
|
||||
}
|
||||
|
||||
albumsStore := albums.NewStore(pool)
|
||||
albumsSvc := albums.NewService(albumsStore, members.NewStore(pool), hub, backend)
|
||||
album, err := albumsSvc.Create(ctx, owner.ID, albums.CreateInput{Title: "Share Album"})
|
||||
if err != nil {
|
||||
t.Fatalf("create album: %v", err)
|
||||
}
|
||||
|
||||
photosSvc := photos.NewService(photos.NewStore(pool), albumsSvc, backend, hub)
|
||||
pngData := makeTestPNG(t)
|
||||
if _, err := photosSvc.Upload(ctx, photos.UploadInput{
|
||||
AlbumID: album.ID, UserID: owner.ID, Filename: "pic.png", Data: pngData,
|
||||
}); err != nil {
|
||||
t.Fatalf("upload: %v", err)
|
||||
}
|
||||
|
||||
var keys []string
|
||||
rows, err := pool.Query(ctx,
|
||||
`SELECT o.storage_key FROM photo_objects po
|
||||
JOIN objects o ON o.id = po.object_id
|
||||
JOIN photos p ON p.id = po.photo_id
|
||||
WHERE p.album_id = $1`, album.ID)
|
||||
if err == nil {
|
||||
for rows.Next() {
|
||||
var k string
|
||||
if rows.Scan(&k) == nil {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM photos WHERE album_id = $1`, album.ID)
|
||||
for _, k := range keys {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM objects WHERE storage_key = $1`, k)
|
||||
}
|
||||
_ = backend.Delete(context.Background(), keys...)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM albums WHERE id = $1`, album.ID)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, owner.ID)
|
||||
})
|
||||
|
||||
shareSvc := sharelinks.NewService(sharelinks.NewStore(pool), albumsSvc)
|
||||
link, err := shareSvc.Create(ctx, album.ID, owner.ID, time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("share create: %v", err)
|
||||
}
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/share/"+link.Token+"/photos", nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("share photos status = %d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Photos []models.Photo `json:"photos"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if len(body.Photos) != 1 {
|
||||
t.Fatalf("photos len = %d, want 1", len(body.Photos))
|
||||
}
|
||||
if body.Photos[0].OriginalURL == "" {
|
||||
t.Fatalf("empty original url")
|
||||
}
|
||||
verify(t, router, body.Photos[0].OriginalURL)
|
||||
}
|
||||
|
||||
func TestShareFlowWithFS(t *testing.T) {
|
||||
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("signer: %v", err)
|
||||
}
|
||||
fs, err := storage.NewFS(t.TempDir(), signer)
|
||||
if err != nil {
|
||||
t.Fatalf("new fs: %v", err)
|
||||
}
|
||||
runShareFlow(t, fs, func(t *testing.T, router *gin.Engine, raw string) {
|
||||
if !strings.Contains(raw, "/api/objects/") {
|
||||
t.Fatalf("expected fs signed url, got %q", raw)
|
||||
}
|
||||
u, err := url.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse url: %v", err)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, u.Path+"?"+u.RawQuery, nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("fs object fetch status = %d", rec.Code)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestShareFlowWithS3(t *testing.T) {
|
||||
endpoint := os.Getenv("S3_ENDPOINT")
|
||||
if endpoint == "" {
|
||||
t.Skip("S3_ENDPOINT not set")
|
||||
}
|
||||
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
|
||||
if err != nil {
|
||||
t.Skipf("signer init: %v", err)
|
||||
}
|
||||
cl, err := storage.NewS3(context.Background(), config.S3{
|
||||
Endpoint: endpoint,
|
||||
Region: os.Getenv("S3_REGION"),
|
||||
AccessKey: os.Getenv("S3_ACCESS_KEY"),
|
||||
SecretKey: os.Getenv("S3_SECRET_KEY"),
|
||||
Bucket: os.Getenv("S3_BUCKET"),
|
||||
UsePathStyle: true,
|
||||
}, signer)
|
||||
if err != nil {
|
||||
t.Skipf("storage init: %v", err)
|
||||
}
|
||||
runShareFlow(t, cl, func(t *testing.T, router *gin.Engine, raw string) {
|
||||
if !strings.Contains(raw, "/api/objects/") {
|
||||
t.Fatalf("expected backend url, got %q", raw)
|
||||
}
|
||||
if strings.Contains(raw, "X-Amz-Signature") {
|
||||
t.Fatalf("presigned url leaked to client: %q", raw)
|
||||
}
|
||||
u, err := url.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse url: %v", err)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, u.Path+"?"+u.RawQuery, nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("s3 object fetch status = %d", rec.Code)
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user