s3 support

internal/app/files/blobfs.go

@@ -27,6 +27,12 @@ type BlobBackend interface {
 	// GetRange 只读 [offset, offset+limit) 段并返回该段字节与文件总大小（limit<=0 读到末尾），
 	// 避免大文件每个 chunk 都整文件读入内存（getFile 按 chunk 多次请求 ⇒ 否则 O(N²) 放大）。
 	GetRange(ctx context.Context, objectKey string, offset, limit int64) (data []byte, total int64, err error)
+	// Delete removes the object at objectKey. Callers (storage retention GC)
+	// must first confirm no remaining file_blobs row on this backend still
+	// references objectKey -- content-addressed storage means the same
+	// object can be shared by multiple documents/photos. Deleting an
+	// already-absent object is not an error.
+	Delete(ctx context.Context, objectKey string) error
 }
 
 // UploadPartBackend 保存 upload.saveFilePart/saveBigFilePart 的临时分片字节。
@@ -205,6 +211,16 @@ func (l *LocalFS) GetRange(_ context.Context, objectKey string, offset, limit in
 	return buf[:read], total, nil
 }
 
+// Delete removes the on-disk object for objectKey. Missing objects are not
+// an error (idempotent, safe to retry). Callers must have already confirmed
+// no other file_blobs row on this backend still references objectKey.
+func (l *LocalFS) Delete(_ context.Context, objectKey string) error {
+	if err := os.Remove(l.pathFor(objectKey)); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("delete blob: %w", err)
+	}
+	return nil
+}
+
 func (l *LocalFS) openBlobFile(objectKey string) (*sharedBlobFile, error) {
 	l.mu.Lock()
 	defer l.mu.Unlock()

internal/app/files/blobfs_test.go

@@ -44,6 +44,32 @@ func TestLocalFSPutGetRoundTrip(t *testing.T) {
 	}
 }
 
+func TestLocalFSDelete(t *testing.T) {
+	fs, err := NewLocalFS(t.TempDir())
+	if err != nil {
+		t.Fatalf("new local fs: %v", err)
+	}
+	ctx := context.Background()
+	key, err := fs.Put(ctx, []byte("delete me"))
+	if err != nil {
+		t.Fatalf("put: %v", err)
+	}
+	if _, err := fs.Get(ctx, key); err != nil {
+		t.Fatalf("get before delete: %v", err)
+	}
+	if err := fs.Delete(ctx, key); err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	if _, err := fs.Get(ctx, key); err == nil {
+		t.Fatal("expected get after delete to fail")
+	}
+	// Deleting an already-absent object must be idempotent, not an error --
+	// the retention sweep can legitimately retry after a partial failure.
+	if err := fs.Delete(ctx, key); err != nil {
+		t.Fatalf("delete already-missing object: %v", err)
+	}
+}
+
 func TestLocalFSPutReaderRoundTrip(t *testing.T) {
 	fs, err := NewLocalFS(t.TempDir())
 	if err != nil {

internal/app/files/blobs3.go

@@ -0,0 +1,179 @@
+package files
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+
+	minio "github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+)
+
+// S3FS stores blob bytes in an S3-compatible object store (self-hosted MinIO
+// or AWS S3). It implements BlobBackend using the same content-addressed
+// sha256-hex object key scheme as LocalFS, so file_blobs.object_key stays
+// comparable regardless of which backend wrote a given row and a deployment
+// can run with rows split across both backends (see TELESRV_BLOB_BACKEND).
+//
+// UploadPartBackend is intentionally NOT implemented here: transient upload
+// parts stay on local disk (see cmd/telesrv/main.go) even when the
+// permanent blob backend is s3 -- one S3 round trip per ~512KB chunk isn't
+// worth it for scratch data that's deleted within minutes of assembly.
+type S3FS struct {
+	client *minio.Client
+	bucket string
+}
+
+// NewS3FS creates an S3-compatible blob backend. endpoint is host[:port]
+// without a scheme (e.g. "minio.internal:9000" or "s3.amazonaws.com");
+// useSSL selects http vs https. pathStyle forces path-style addressing
+// (bucket in the URL path rather than as a subdomain), which self-hosted
+// MinIO typically requires and AWS S3 does not.
+func NewS3FS(ctx context.Context, endpoint, accessKeyID, secretAccessKey, bucket, region string, useSSL, pathStyle bool) (*S3FS, error) {
+	if endpoint == "" || bucket == "" {
+		return nil, fmt.Errorf("s3 blob backend: endpoint and bucket are required")
+	}
+	lookup := minio.BucketLookupAuto
+	if pathStyle {
+		lookup = minio.BucketLookupPath
+	}
+	client, err := minio.New(endpoint, &minio.Options{
+		Creds:        credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
+		Secure:       useSSL,
+		Region:       region,
+		BucketLookup: lookup,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("create s3 client: %w", err)
+	}
+	exists, err := client.BucketExists(ctx, bucket)
+	if err != nil {
+		return nil, fmt.Errorf("check s3 bucket %q: %w", bucket, err)
+	}
+	if !exists {
+		if err := client.MakeBucket(ctx, bucket, minio.MakeBucketOptions{Region: region}); err != nil {
+			return nil, fmt.Errorf("create s3 bucket %q: %w", bucket, err)
+		}
+	}
+	return &S3FS{client: client, bucket: bucket}, nil
+}
+
+// Name 返回后端标识，与 file_blobs.backend 一致。
+func (s *S3FS) Name() string { return "s3" }
+
+func (s *S3FS) Put(ctx context.Context, data []byte) (string, error) {
+	key, _, _, err := s.PutReader(ctx, bytes.NewReader(data))
+	return key, err
+}
+
+// PutReader hashes the stream to a local temp file first (so the sha256 key
+// and exact size are known before the S3 PUT, matching LocalFS's
+// content-addressed dedup semantics -- an unknown-length streaming PUT would
+// need a second copy operation to rename-by-hash after the fact, which S3
+// has no equivalent of), then uploads it and checks for an existing object
+// with that key first to skip a redundant PUT.
+func (s *S3FS) PutReader(ctx context.Context, r io.Reader) (string, int64, []byte, error) {
+	tmp, err := os.CreateTemp("", "blob-s3-*.tmp")
+	if err != nil {
+		return "", 0, nil, fmt.Errorf("create s3 blob staging file: %w", err)
+	}
+	tmpPath := tmp.Name()
+	defer os.Remove(tmpPath)
+
+	h := sha256.New()
+	size, err := copyWithContext(ctx, io.MultiWriter(tmp, h), r)
+	closeErr := tmp.Close()
+	if err != nil {
+		return "", 0, nil, fmt.Errorf("stage s3 blob: %w", err)
+	}
+	if closeErr != nil {
+		return "", 0, nil, fmt.Errorf("close s3 blob staging file: %w", closeErr)
+	}
+	sum := h.Sum(nil)
+	key := hex.EncodeToString(sum)
+
+	if _, err := s.client.StatObject(ctx, s.bucket, key, minio.StatObjectOptions{}); err == nil {
+		return key, size, append([]byte(nil), sum...), nil
+	} else if !isS3NotFound(err) {
+		return "", 0, nil, fmt.Errorf("stat s3 blob: %w", err)
+	}
+
+	f, err := os.Open(tmpPath)
+	if err != nil {
+		return "", 0, nil, fmt.Errorf("reopen s3 blob staging file: %w", err)
+	}
+	defer f.Close()
+	if _, err := s.client.PutObject(ctx, s.bucket, key, f, size, minio.PutObjectOptions{}); err != nil {
+		return "", 0, nil, fmt.Errorf("put s3 blob: %w", err)
+	}
+	return key, size, append([]byte(nil), sum...), nil
+}
+
+func (s *S3FS) Get(ctx context.Context, objectKey string) ([]byte, error) {
+	obj, err := s.client.GetObject(ctx, s.bucket, objectKey, minio.GetObjectOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("get s3 blob: %w", err)
+	}
+	defer obj.Close()
+	data, err := io.ReadAll(obj)
+	if err != nil {
+		return nil, fmt.Errorf("read s3 blob: %w", err)
+	}
+	return data, nil
+}
+
+// GetRange 语义与 LocalFS.GetRange 一致：只读 [offset, offset+limit) 段，limit<=0 读到末尾，
+// total 取自对象实际大小。
+func (s *S3FS) GetRange(ctx context.Context, objectKey string, offset, limit int64) ([]byte, int64, error) {
+	info, err := s.client.StatObject(ctx, s.bucket, objectKey, minio.StatObjectOptions{})
+	if err != nil {
+		return nil, 0, fmt.Errorf("stat s3 blob: %w", err)
+	}
+	total := info.Size
+	if offset < 0 {
+		offset = 0
+	}
+	if offset >= total {
+		return []byte{}, total, nil
+	}
+	opts := minio.GetObjectOptions{}
+	end := total - 1
+	if limit > 0 && offset+limit-1 < end {
+		end = offset + limit - 1
+	}
+	if err := opts.SetRange(offset, end); err != nil {
+		return nil, 0, fmt.Errorf("set s3 range: %w", err)
+	}
+	obj, err := s.client.GetObject(ctx, s.bucket, objectKey, opts)
+	if err != nil {
+		return nil, 0, fmt.Errorf("get s3 blob range: %w", err)
+	}
+	defer obj.Close()
+	data, err := io.ReadAll(obj)
+	if err != nil {
+		return nil, 0, fmt.Errorf("read s3 blob range: %w", err)
+	}
+	return data, total, nil
+}
+
+// Delete removes the object at objectKey. A missing object is not an error
+// (idempotent, safe to retry). Callers must have already confirmed no other
+// file_blobs row on this backend still references objectKey.
+func (s *S3FS) Delete(ctx context.Context, objectKey string) error {
+	if err := s.client.RemoveObject(ctx, s.bucket, objectKey, minio.RemoveObjectOptions{}); err != nil {
+		if isS3NotFound(err) {
+			return nil
+		}
+		return fmt.Errorf("delete s3 blob: %w", err)
+	}
+	return nil
+}
+
+func isS3NotFound(err error) bool {
+	resp := minio.ToErrorResponse(err)
+	return resp.Code == "NoSuchKey" || resp.Code == "NotFound" || resp.StatusCode == 404
+}

internal/app/files/diskspace.go

@@ -0,0 +1,99 @@
+package files
+
+import "sync/atomic"
+
+// SpaceGuard bounds how much more may be written to the permanent blob
+// backend. LocalDiskSpaceGuard checks real OS free disk bytes;
+// S3BudgetSpaceGuard compares a cached tracked-bytes total against a
+// configured budget (S3 has no OS-level "free space" concept). Both are
+// refreshed periodically by DiskUsageWorker rather than recomputed on every
+// upload chunk.
+type SpaceGuard interface {
+	// Allow reports whether writing `additional` more bytes is currently
+	// permitted. false should surface to the client as domain.ErrStorageFull.
+	Allow(additional int64) (bool, error)
+	// Usage returns the last-refreshed (used, total) byte snapshot for the
+	// admin panel. ok is false if no successful refresh has happened yet.
+	Usage() (used, total int64, ok bool)
+}
+
+// NoopSpaceGuard always allows writes; the default when the low-space guard
+// is disabled by configuration.
+type NoopSpaceGuard struct{}
+
+func (NoopSpaceGuard) Allow(int64) (bool, error)  { return true, nil }
+func (NoopSpaceGuard) Usage() (int64, int64, bool) { return 0, 0, false }
+
+// LocalDiskSpaceGuard rejects writes once cached free disk bytes fall below
+// minFreeBytes (<=0 disables the check). The free-bytes figure is
+// refreshed by DiskUsageWorker, not recomputed per call, to avoid a statfs
+// syscall on every upload chunk -- reads are lock-free.
+type LocalDiskSpaceGuard struct {
+	minFreeBytes int64
+	free         atomic.Int64
+	total        atomic.Int64
+	ready        atomic.Bool
+}
+
+func NewLocalDiskSpaceGuard(minFreeBytes int64) *LocalDiskSpaceGuard {
+	return &LocalDiskSpaceGuard{minFreeBytes: minFreeBytes}
+}
+
+func (g *LocalDiskSpaceGuard) Allow(additional int64) (bool, error) {
+	// Before the first refresh completes, allow rather than reject: a
+	// startup race shouldn't turn into spurious upload failures.
+	if g.minFreeBytes <= 0 || !g.ready.Load() {
+		return true, nil
+	}
+	return g.free.Load()-additional >= g.minFreeBytes, nil
+}
+
+func (g *LocalDiskSpaceGuard) Usage() (used, total int64, ok bool) {
+	if !g.ready.Load() {
+		return 0, 0, false
+	}
+	total = g.total.Load()
+	used = total - g.free.Load()
+	if used < 0 {
+		used = 0
+	}
+	return used, total, true
+}
+
+func (g *LocalDiskSpaceGuard) setFree(free, total int64) {
+	g.free.Store(free)
+	g.total.Store(total)
+	g.ready.Store(true)
+}
+
+// S3BudgetSpaceGuard rejects writes once a cached tracked-bytes total
+// (refreshed periodically from file_blobs) would exceed maxTotalBytes
+// (<=0 disables the check).
+type S3BudgetSpaceGuard struct {
+	maxTotalBytes int64
+	used          atomic.Int64
+	ready         atomic.Bool
+}
+
+func NewS3BudgetSpaceGuard(maxTotalBytes int64) *S3BudgetSpaceGuard {
+	return &S3BudgetSpaceGuard{maxTotalBytes: maxTotalBytes}
+}
+
+func (g *S3BudgetSpaceGuard) Allow(additional int64) (bool, error) {
+	if g.maxTotalBytes <= 0 || !g.ready.Load() {
+		return true, nil
+	}
+	return g.used.Load()+additional <= g.maxTotalBytes, nil
+}
+
+func (g *S3BudgetSpaceGuard) Usage() (used, total int64, ok bool) {
+	if !g.ready.Load() {
+		return 0, 0, false
+	}
+	return g.used.Load(), g.maxTotalBytes, true
+}
+
+func (g *S3BudgetSpaceGuard) setUsed(used int64) {
+	g.used.Store(used)
+	g.ready.Store(true)
+}

internal/app/files/diskspace_test.go

@@ -0,0 +1,76 @@
+package files
+
+import "testing"
+
+func TestNoopSpaceGuardAlwaysAllows(t *testing.T) {
+	g := NoopSpaceGuard{}
+	if allowed, err := g.Allow(1 << 40); err != nil || !allowed {
+		t.Fatalf("expected noop guard to always allow, got allowed=%v err=%v", allowed, err)
+	}
+	if _, _, ok := g.Usage(); ok {
+		t.Fatal("expected noop guard to report no usage snapshot")
+	}
+}
+
+func TestLocalDiskSpaceGuardBeforeFirstRefresh(t *testing.T) {
+	g := NewLocalDiskSpaceGuard(1 << 30)
+	// Before setFree has ever run, the guard must not reject -- a startup
+	// race shouldn't turn into spurious upload failures.
+	if allowed, err := g.Allow(1 << 40); err != nil || !allowed {
+		t.Fatalf("expected pre-refresh guard to allow, got allowed=%v err=%v", allowed, err)
+	}
+	if _, _, ok := g.Usage(); ok {
+		t.Fatal("expected no usage snapshot before first refresh")
+	}
+}
+
+func TestLocalDiskSpaceGuardThreshold(t *testing.T) {
+	const minFree = int64(1000)
+	g := NewLocalDiskSpaceGuard(minFree)
+	g.setFree(1500, 10000)
+
+	if allowed, err := g.Allow(400); err != nil || !allowed {
+		t.Fatalf("writing 400 bytes leaves 1100 free (>= 1000 min): want allow, got allowed=%v err=%v", allowed, err)
+	}
+	if allowed, err := g.Allow(600); err != nil || allowed {
+		t.Fatalf("writing 600 bytes leaves 900 free (< 1000 min): want reject, got allowed=%v err=%v", allowed, err)
+	}
+
+	used, total, ok := g.Usage()
+	if !ok || total != 10000 || used != 8500 {
+		t.Fatalf("usage snapshot = used=%d total=%d ok=%v, want used=8500 total=10000 ok=true", used, total, ok)
+	}
+}
+
+func TestLocalDiskSpaceGuardDisabled(t *testing.T) {
+	g := NewLocalDiskSpaceGuard(0)
+	g.setFree(10, 1000)
+	if allowed, err := g.Allow(1 << 40); err != nil || !allowed {
+		t.Fatalf("minFreeBytes<=0 must disable the check, got allowed=%v err=%v", allowed, err)
+	}
+}
+
+func TestS3BudgetSpaceGuardThreshold(t *testing.T) {
+	const maxTotal = int64(10000)
+	g := NewS3BudgetSpaceGuard(maxTotal)
+	g.setUsed(9000)
+
+	if allowed, err := g.Allow(1000); err != nil || !allowed {
+		t.Fatalf("9000+1000 == budget: want allow, got allowed=%v err=%v", allowed, err)
+	}
+	if allowed, err := g.Allow(1001); err != nil || allowed {
+		t.Fatalf("9000+1001 exceeds budget: want reject, got allowed=%v err=%v", allowed, err)
+	}
+
+	used, total, ok := g.Usage()
+	if !ok || total != maxTotal || used != 9000 {
+		t.Fatalf("usage snapshot = used=%d total=%d ok=%v, want used=9000 total=%d ok=true", used, total, ok, maxTotal)
+	}
+}
+
+func TestS3BudgetSpaceGuardBeforeFirstRefresh(t *testing.T) {
+	g := NewS3BudgetSpaceGuard(100)
+	if allowed, err := g.Allow(1 << 40); err != nil || !allowed {
+		t.Fatalf("expected pre-refresh guard to allow, got allowed=%v err=%v", allowed, err)
+	}
+}

internal/app/files/diskspace_unix.go

@@ -0,0 +1,15 @@
+//go:build !windows
+
+package files
+
+import "golang.org/x/sys/unix"
+
+// localDiskFreeBytes returns free (available to an unprivileged writer, not
+// counting reserved blocks) and total bytes for the filesystem containing path.
+func localDiskFreeBytes(path string) (free, total int64, err error) {
+	var st unix.Statfs_t
+	if err := unix.Statfs(path, &st); err != nil {
+		return 0, 0, err
+	}
+	return int64(st.Bavail) * int64(st.Bsize), int64(st.Blocks) * int64(st.Bsize), nil
+}

internal/app/files/diskspace_windows.go

@@ -0,0 +1,19 @@
+//go:build windows
+
+package files
+
+import "golang.org/x/sys/windows"
+
+// localDiskFreeBytes returns free (available to the calling user) and total
+// bytes for the volume containing path.
+func localDiskFreeBytes(path string) (free, total int64, err error) {
+	ptr, err := windows.UTF16PtrFromString(path)
+	if err != nil {
+		return 0, 0, err
+	}
+	var freeAvail, totalBytes, totalFree uint64
+	if err := windows.GetDiskFreeSpaceEx(ptr, &freeAvail, &totalBytes, &totalFree); err != nil {
+		return 0, 0, err
+	}
+	return int64(freeAvail), int64(totalBytes), nil
+}

internal/app/files/diskusage_worker.go

@@ -0,0 +1,81 @@
+package files
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+// SumFileBlobBytesStore is the minimal store dependency DiskUsageWorker
+// needs to refresh an S3BudgetSpaceGuard.
+type SumFileBlobBytesStore interface {
+	SumFileBlobBytes(ctx context.Context) (int64, error)
+}
+
+// DiskUsageWorker periodically refreshes one SpaceGuard's cached usage
+// snapshot, so the upload path never pays a statfs syscall / SUM(size)
+// query per chunk.
+type DiskUsageWorker struct {
+	interval time.Duration
+	log      *zap.Logger
+	refresh  func(ctx context.Context) error
+}
+
+// NewLocalDiskUsageWorker refreshes a LocalDiskSpaceGuard from real OS free
+// disk bytes under root (the blob backend's storage directory).
+func NewLocalDiskUsageWorker(guard *LocalDiskSpaceGuard, root string, interval time.Duration, log *zap.Logger) *DiskUsageWorker {
+	return newDiskUsageWorker(interval, log, func(context.Context) error {
+		free, total, err := localDiskFreeBytes(root)
+		if err != nil {
+			return err
+		}
+		guard.setFree(free, total)
+		return nil
+	})
+}
+
+// NewS3DiskUsageWorker refreshes an S3BudgetSpaceGuard from the tracked
+// file_blobs byte total.
+func NewS3DiskUsageWorker(guard *S3BudgetSpaceGuard, media SumFileBlobBytesStore, interval time.Duration, log *zap.Logger) *DiskUsageWorker {
+	return newDiskUsageWorker(interval, log, func(ctx context.Context) error {
+		used, err := media.SumFileBlobBytes(ctx)
+		if err != nil {
+			return err
+		}
+		guard.setUsed(used)
+		return nil
+	})
+}
+
+func newDiskUsageWorker(interval time.Duration, log *zap.Logger, refresh func(context.Context) error) *DiskUsageWorker {
+	if interval <= 0 {
+		interval = time.Minute
+	}
+	if log == nil {
+		log = zap.NewNop()
+	}
+	return &DiskUsageWorker{interval: interval, log: log, refresh: refresh}
+}
+
+// Run refreshes once immediately (so the guard isn't stuck "not ready" for
+// a full interval after startup), then on every tick until ctx is done.
+func (w *DiskUsageWorker) Run(ctx context.Context) {
+	w.refreshOnce(ctx)
+	ticker := time.NewTicker(w.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			w.refreshOnce(ctx)
+		}
+	}
+}
+
+func (w *DiskUsageWorker) refreshOnce(ctx context.Context) {
+	if err := w.refresh(ctx); err != nil {
+		w.log.Warn("refresh storage usage snapshot failed", zap.Error(err))
+	}
+}

internal/app/files/photos.go

@@ -63,7 +63,7 @@ func (s *Service) CreatePhotoFromUpload(ctx context.Context, file domain.Uploade
 	if len(data) == 0 {
 		return domain.Photo{}, domain.ErrPhotoInvalid
 	}
-	photo, err := s.createPhoto(ctx, data, photoSizeSpecsForMessage(data))
+	photo, err := s.createPhoto(ctx, data, photoSizeSpecsForMessage(data), file.OwnerUserID)
 	if err != nil {
 		return domain.Photo{}, err
 	}
@@ -86,11 +86,13 @@ func (s *Service) CreatePhotoFromUpload(ctx context.Context, file domain.Uploade
 }
 
 // CreatePhotoFromBytes stores already-fetched image bytes as a message Photo.
+// There is no uploader (e.g. a server-fetched webpage preview image), so the
+// photo is attributed to the system (owner_user_id 0) for storage accounting.
 func (s *Service) CreatePhotoFromBytes(ctx context.Context, data []byte) (domain.Photo, error) {
 	if len(data) == 0 {
 		return domain.Photo{}, domain.ErrPhotoInvalid
 	}
-	return s.createPhoto(ctx, data, photoSizeSpecsForMessage(data))
+	return s.createPhoto(ctx, data, photoSizeSpecsForMessage(data), 0)
 }
 
 // GetPhoto 按 id 返回已存储照片。
@@ -148,7 +150,7 @@ func (s *Service) CreateAvatarFromUpload(ctx context.Context, file domain.Upload
 	if len(data) == 0 {
 		return domain.Photo{}, domain.ErrPhotoInvalid
 	}
-	return s.createAvatarPhoto(ctx, data)
+	return s.createAvatarPhoto(ctx, data, file.OwnerUserID)
 }
 
 // CreateAvatarVideoFromUpload stores an animated profile video as photo.video_sizes.
@@ -206,6 +208,7 @@ func (s *Service) createAvatarVideoFromUpload(ctx context.Context, file domain.U
 		Date:          int(time.Now().Unix()),
 		DCID:          s.dc,
 		Sizes:         sizes,
+		OwnerUserID:   file.OwnerUserID,
 	}
 	if err := s.media.PutPhoto(ctx, photo); err != nil {
 		return domain.Photo{}, err
@@ -306,6 +309,7 @@ func (s *Service) CreateDocumentFromUpload(ctx context.Context, file domain.Uplo
 		Size:          body.Size,
 		DCID:          s.dc,
 		Attributes:    spec.Attributes,
+		OwnerUserID:   file.OwnerUserID,
 	}
 	thumbMaterialized := false
 	if spec.Thumb != nil {
@@ -620,7 +624,7 @@ func (s *Service) DeleteProfilePhotosKind(ctx context.Context, ownerType domain.
 }
 
 // createPhoto 把字节落 blob（每个尺寸一个 location_key，指向同一内容）并写 photos 表。
-func (s *Service) createPhoto(ctx context.Context, data []byte, specs []photoSizeSpec) (domain.Photo, error) {
+func (s *Service) createPhoto(ctx context.Context, data []byte, specs []photoSizeSpec, ownerUserID int64) (domain.Photo, error) {
 	photoID := randomID()
 	sizes, err := s.putPhotoStaticSizes(ctx, photoID, data, specs)
 	if err != nil {
@@ -633,6 +637,7 @@ func (s *Service) createPhoto(ctx context.Context, data []byte, specs []photoSiz
 		Date:          int(time.Now().Unix()),
 		DCID:          s.dc,
 		Sizes:         sizes,
+		OwnerUserID:   ownerUserID,
 	}
 	if err := s.media.PutPhoto(ctx, photo); err != nil {
 		return domain.Photo{}, err
@@ -640,7 +645,7 @@ func (s *Service) createPhoto(ctx context.Context, data []byte, specs []photoSiz
 	return photo, nil
 }
 
-func (s *Service) createAvatarPhoto(ctx context.Context, data []byte) (domain.Photo, error) {
+func (s *Service) createAvatarPhoto(ctx context.Context, data []byte, ownerUserID int64) (domain.Photo, error) {
 	photoID := randomID()
 	sizes, err := s.putAvatarStaticSizes(ctx, photoID, data, photoSizeSpecsForAvatar(data))
 	if err != nil {
@@ -653,6 +658,7 @@ func (s *Service) createAvatarPhoto(ctx context.Context, data []byte) (domain.Ph
 		Date:          int(time.Now().Unix()),
 		DCID:          s.dc,
 		Sizes:         sizes,
+		OwnerUserID:   ownerUserID,
 	}
 	if err := s.media.PutPhoto(ctx, photo); err != nil {
 		return domain.Photo{}, err

internal/app/files/retention.go

@@ -0,0 +1,94 @@
+package files
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.uber.org/zap"
+
+	"telesrv/internal/domain"
+)
+
+// mediaRetentionStore is implemented by store.MediaStore backends that
+// support the storage retention sweep (currently only the Postgres store).
+// A type assertion, not a MediaStore interface method, keeps these
+// admin/maintenance-only queries out of the hot RPC-facing interface --
+// same convention as photoBatchStore above.
+type mediaRetentionStore interface {
+	ListOrphanedDocumentIDsOlderThan(ctx context.Context, cutoff time.Time, limit int) ([]int64, error)
+	ListOrphanedPhotoIDsOlderThan(ctx context.Context, cutoff time.Time, limit int) ([]int64, error)
+	CountFileBlobRefs(ctx context.Context, backend, objectKey string) (int, error)
+	DeleteDocumentAndBlobs(ctx context.Context, id int64) ([]domain.FileBlob, error)
+	DeletePhotoAndBlobs(ctx context.Context, id int64) ([]domain.FileBlob, error)
+}
+
+// DeleteOrphanedOlderThan implements maintenance.OrphanedMediaRetentionStore:
+// permanently deletes documents/photos that have had no live reference
+// (message/profile-photo/sticker-set, see media_references) since at least
+// cutoff, along with their blob(s) -- but only physically removes bytes
+// from the backend once confirming no other file_blobs row still needs the
+// object, since content-addressed storage means the same bytes can be
+// shared across documents/photos.
+func (s *Service) DeleteOrphanedOlderThan(ctx context.Context, cutoff time.Time, limit int) (int, error) {
+	store, ok := s.media.(mediaRetentionStore)
+	if !ok || limit <= 0 {
+		return 0, nil
+	}
+	deleted := 0
+	docIDs, err := store.ListOrphanedDocumentIDsOlderThan(ctx, cutoff, limit)
+	if err != nil {
+		return deleted, fmt.Errorf("list orphaned documents: %w", err)
+	}
+	for _, id := range docIDs {
+		blobs, err := store.DeleteDocumentAndBlobs(ctx, id)
+		if err != nil {
+			s.log.Warn("delete orphaned document failed", zap.Int64("document_id", id), zap.Error(err))
+			continue
+		}
+		s.deleteOrphanedBlobs(ctx, store, blobs)
+		deleted++
+	}
+	photoIDs, err := store.ListOrphanedPhotoIDsOlderThan(ctx, cutoff, limit)
+	if err != nil {
+		return deleted, fmt.Errorf("list orphaned photos: %w", err)
+	}
+	for _, id := range photoIDs {
+		blobs, err := store.DeletePhotoAndBlobs(ctx, id)
+		if err != nil {
+			s.log.Warn("delete orphaned photo failed", zap.Int64("photo_id", id), zap.Error(err))
+			continue
+		}
+		s.deleteOrphanedBlobs(ctx, store, blobs)
+		deleted++
+	}
+	return deleted, nil
+}
+
+// deleteOrphanedBlobs removes each blob from its backend once confirming
+// (via CountFileBlobRefs) no other file_blobs row still references
+// (backend, object_key). Only blobs on the currently active backend are
+// physically removed -- a blob left over from a previously active backend
+// (deployment switched TELESRV_BLOB_BACKEND at some point; switching back
+// isn't supported) is logged and skipped rather than silently dropped,
+// since there's no configured client to reach it right now anyway.
+func (s *Service) deleteOrphanedBlobs(ctx context.Context, store mediaRetentionStore, blobs []domain.FileBlob) {
+	for _, b := range blobs {
+		refs, err := store.CountFileBlobRefs(ctx, string(b.Backend), b.ObjectKey)
+		if err != nil {
+			s.log.Warn("count file blob refs failed", zap.String("object_key", b.ObjectKey), zap.Error(err))
+			continue
+		}
+		if refs > 0 {
+			continue
+		}
+		if string(b.Backend) != s.blobs.Name() {
+			s.log.Warn("orphaned blob is on an inactive backend, skipping physical delete",
+				zap.String("backend", string(b.Backend)), zap.String("object_key", b.ObjectKey))
+			continue
+		}
+		if err := s.blobs.Delete(ctx, b.ObjectKey); err != nil {
+			s.log.Warn("delete orphaned blob failed", zap.String("object_key", b.ObjectKey), zap.Error(err))
+		}
+	}
+}

internal/app/files/seed_test.go

@@ -168,6 +168,16 @@ func (f *fakeMediaStore) GetFileBlobs(_ context.Context, keys []string) (map[str
 	return out, nil
 }
 
+func (f *fakeMediaStore) SumFileBlobBytes(_ context.Context) (int64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var total int64
+	for _, b := range f.blobs {
+		total += b.Size
+	}
+	return total, nil
+}
+
 func (f *fakeMediaStore) GetSeedState(_ context.Context, key string) (string, bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()

internal/app/files/service.go

@@ -64,6 +64,7 @@ type Service struct {
 	stickerSetCache    *stickerSetFullCache
 	stickerSetNegCache *stickerSetNegativeCache
 	uploadQuota        domain.UploadPartQuota
+	spaceGuard         SpaceGuard
 	mapTiles           *mapTileProxy
 	externalMedia      *externalMediaFetcher
 	webpage            *webpageFetcher
@@ -116,6 +117,30 @@ func WithUploadPartQuota(quota domain.UploadPartQuota) Option {
 	}
 }
 
+// WithSpaceGuard installs the low-disk-space upload guard. Not calling this
+// (or passing nil) leaves the default NoopSpaceGuard, which never rejects.
+func WithSpaceGuard(guard SpaceGuard) Option {
+	return func(s *Service) {
+		if guard != nil {
+			s.spaceGuard = guard
+		}
+	}
+}
+
+// WithUploadPartBackend overrides where transient upload-part chunks are
+// staged before assembly, independent of the permanent blob backend passed
+// to NewService. Needed when the permanent backend is s3 (S3FS doesn't
+// implement UploadPartBackend -- chunk-per-request S3 round trips aren't
+// worth it for scratch data deleted within minutes): pass a LocalFS here so
+// uploads keep working, while permanent blobs still land in s3.
+func WithUploadPartBackend(backend UploadPartBackend) Option {
+	return func(s *Service) {
+		if backend != nil {
+			s.uploadParts = backend
+		}
+	}
+}
+
 // NewService 创建 files 服务。dc 是本 server 的 DC id，写入新建 document/photo 的 dc_id。
 func NewService(media store.MediaStore, blobs BlobBackend, dc int, opts ...Option) *Service {
 	s := &Service{
@@ -132,6 +157,7 @@ func NewService(media store.MediaStore, blobs BlobBackend, dc int, opts ...Optio
 			MaxParts: DefaultUploadInFlightMaxParts,
 			MaxFiles: DefaultUploadInFlightMaxFiles,
 		},
+		spaceGuard: NoopSpaceGuard{},
 	}
 	if partBackend, ok := blobs.(UploadPartBackend); ok {
 		s.uploadParts = partBackend
@@ -205,6 +231,16 @@ func (s *Service) saveFilePart(ctx context.Context, part domain.UploadPart, byte
 	if s.uploadParts == nil {
 		return fmt.Errorf("upload part backend not configured")
 	}
+	// Cheapest possible rejection point: reject before any disk write once
+	// the permanent blob backend is low on space. Upload parts themselves
+	// always land on local scratch disk (see UploadPartBackend), but a
+	// low-space condition on the permanent backend means assembly will
+	// fail anyway, so there's no point accepting more chunks toward it.
+	if allowed, err := s.spaceGuard.Allow(int64(len(bytes))); err != nil {
+		return err
+	} else if !allowed {
+		return domain.ErrStorageFull
+	}
 	slot, err := s.checkUploadPartQuota(ctx, part)
 	if err != nil {
 		return err
@@ -584,10 +620,19 @@ type assembledUploadBlob struct {
 // assembleUploadBlob 把上传分片流式写入正式 blob。调用方应在 durable media 元数据
 // 成功提交后调用 cleanupUploadParts，避免 metadata 写失败时丢失可重试的上传分片。
 func (s *Service) assembleUploadBlob(ctx context.Context, ownerUserID, fileID int64, expectedParts int) (assembledUploadBlob, error) {
-	parts, _, err := s.loadAndValidateUploadParts(ctx, ownerUserID, fileID, expectedParts)
+	parts, total, err := s.loadAndValidateUploadParts(ctx, ownerUserID, fileID, expectedParts)
 	if err != nil {
 		return assembledUploadBlob{}, err
 	}
+	// Re-check free space against the full assembled size right before
+	// committing to the permanent blob backend: SaveFilePart already
+	// checked each chunk, but free space may have dropped since then over
+	// the lifetime of a large multi-part upload.
+	if allowed, err := s.spaceGuard.Allow(total); err != nil {
+		return assembledUploadBlob{}, err
+	} else if !allowed {
+		return assembledUploadBlob{}, domain.ErrStorageFull
+	}
 	if s.uploadParts == nil {
 		return assembledUploadBlob{}, fmt.Errorf("upload part backend not configured")
 	}

internal/app/maintenance/media_retention.go

@@ -0,0 +1,27 @@
+package maintenance
+
+import (
+	"context"
+	"time"
+)
+
+// OrphanedMediaRetentionStore deletes documents/photos that have been
+// orphaned (no live message/profile-photo/sticker-set reference remains,
+// tracked via media_references + orphaned_at) for at least the configured
+// age, along with their underlying blob once no other file_blobs row on
+// its backend still needs it. Never touches media that still has a live
+// reference, regardless of age.
+type OrphanedMediaRetentionStore interface {
+	DeleteOrphanedOlderThan(ctx context.Context, cutoff time.Time, limit int) (int, error)
+}
+
+// WithOrphanedMediaRetention enables the storage retention sweep. maxAge is
+// how long a document/photo must have been orphaned before it's actually
+// deleted (not how old the media itself is) -- <=0 leaves the sweep
+// disabled even if a store is provided, matching
+// TELESRV_STORAGE_RETENTION_ENABLE=false being the safe default.
+func (w *RetentionWorker) WithOrphanedMediaRetention(store OrphanedMediaRetentionStore, maxAge time.Duration) *RetentionWorker {
+	w.orphanedMedia = store
+	w.orphanedMediaMaxAge = maxAge
+	return w
+}

internal/app/maintenance/retention.go

@@ -118,6 +118,7 @@ type RetentionWorker struct {
 	orphanAuthKeys              OrphanAuthKeyRetentionStore
 	activeAuthKeys              ActiveRawAuthKeyProvider
 	activeAuthKeyHeartbeat      ActiveAuthKeyHeartbeatStore
+	orphanedMedia               OrphanedMediaRetentionStore
 	logger                      *zap.Logger
 	retention                   time.Duration
 	botAPIRetention             time.Duration
@@ -126,6 +127,7 @@ type RetentionWorker struct {
 	authDeliveryReportRetention time.Duration
 	outboxPoisonRetention       time.Duration
 	outboxPoisonInterval        time.Duration
+	orphanedMediaMaxAge         time.Duration
 	interval                    time.Duration
 	batch                       int
 }
@@ -412,6 +414,17 @@ func (w *RetentionWorker) runRetentionOnce(ctx context.Context) {
 			w.logger.Info("expired channel_update_events contiguous-prefix cleanup complete", zap.Int("deleted", channelDeleted))
 		}
 	}
+	if w.orphanedMedia != nil && w.orphanedMediaMaxAge > 0 {
+		// Only ever touches documents/photos already marked orphaned (no live
+		// message/profile-photo/sticker-set reference remains) -- media still
+		// visible in a conversation is never a candidate, regardless of age.
+		mediaDeleted, err := w.orphanedMedia.DeleteOrphanedOlderThan(ctx, time.Now().Add(-w.orphanedMediaMaxAge), w.batch)
+		if err != nil {
+			w.logger.Warn("orphaned media storage retention sweep failed", zap.Error(err))
+		} else if mediaDeleted > 0 {
+			w.logger.Info("orphaned media storage retention sweep complete", zap.Int("deleted", mediaDeleted))
+		}
+	}
 }
 
 func (w *RetentionWorker) orphanHeartbeatInterval() time.Duration {