perf(api): async S3 uploads in StoreMessage (#471)

* perf(api): async S3 uploads in StoreMessage to reduce latency

S3 upload latency was the bottleneck in StoreMessage (p75 ~1.8s).
Pre-compute asset metadata (SHA256, S3 key) in-memory, write DB + Redis
cache synchronously, then upload to S3 in a background goroutine.
This also fixes context canceled errors when clients disconnect early.

* test(api): add unit tests for PrepareJSONAsset and PrepareFormFileAsset

Author: GenerQAQ

src/server/api/go/internal/infra/blob/s3.go

@@ -353,6 +353,98 @@ func (u *S3Deps) UploadJSON(ctx context.Context, keyPrefix string, data interfac
 	)
 }
 
+// PreparedUpload holds pre-computed asset metadata and content for deferred S3 upload.
+type PreparedUpload struct {
+	Asset    model.Asset       // Pre-computed asset metadata (S3Key, SHA256, MIME, SizeB)
+	Content  []byte            // Serialized content to upload
+	Metadata map[string]string // S3 object metadata
+}
+
+// PrepareJSONAsset pre-computes asset metadata for JSON data without making any S3 calls.
+// Returns a PreparedUpload that can later be uploaded via UploadPrepared.
+func (u *S3Deps) PrepareJSONAsset(keyPrefix string, data interface{}) (*PreparedUpload, error) {
+	jsonData, err := sonic.Marshal(data)
+	if err != nil {
+		return nil, fmt.Errorf("marshal json: %w", err)
+	}
+
+	h := sha256.New()
+	h.Write(jsonData)
+	sumHex := hex.EncodeToString(h.Sum(nil))
+
+	datePrefix := time.Now().UTC().Format("2006/01/02")
+	key := fmt.Sprintf("%s/%s/%s.json", keyPrefix, datePrefix, sumHex)
+
+	return &PreparedUpload{
+		Asset: model.Asset{
+			Bucket: u.Bucket,
+			S3Key:  key,
+			SHA256: sumHex,
+			MIME:   "application/json",
+			SizeB:  int64(len(jsonData)),
+		},
+		Content:  jsonData,
+		Metadata: map[string]string{"sha256": sumHex},
+	}, nil
+}
+
+// PrepareFormFileAsset pre-computes asset metadata for a multipart file without making any S3 calls.
+// Returns a PreparedUpload that can later be uploaded via UploadPrepared.
+func (u *S3Deps) PrepareFormFileAsset(keyPrefix string, fh *multipart.FileHeader) (*PreparedUpload, error) {
+	file, err := fh.Open()
+	if err != nil {
+		return nil, err
+	}
+	defer file.Close()
+
+	var buf bytes.Buffer
+	if _, err := io.Copy(&buf, file); err != nil {
+		return nil, err
+	}
+	fileContent := buf.Bytes()
+
+	h := sha256.New()
+	h.Write(fileContent)
+	sumHex := hex.EncodeToString(h.Sum(nil))
+
+	ext := strings.ToLower(filepath.Ext(fh.Filename))
+	contentType := mime.DetectMimeType(fileContent, fh.Filename)
+
+	datePrefix := time.Now().UTC().Format("2006/01/02")
+	key := fmt.Sprintf("%s/%s/%s%s", keyPrefix, datePrefix, sumHex, ext)
+
+	return &PreparedUpload{
+		Asset: model.Asset{
+			Bucket: u.Bucket,
+			S3Key:  key,
+			SHA256: sumHex,
+			MIME:   contentType,
+			SizeB:  int64(len(fileContent)),
+		},
+		Content:  fileContent,
+		Metadata: map[string]string{"sha256": sumHex, "name": fh.Filename},
+	}, nil
+}
+
+// UploadPrepared executes a deferred S3 upload for a previously prepared asset.
+// It uploads directly using the pre-computed S3 key. Since the key contains the
+// content SHA256, re-uploading identical content is naturally idempotent.
+func (u *S3Deps) UploadPrepared(ctx context.Context, p *PreparedUpload) error {
+	input := &s3.PutObjectInput{
+		Bucket:      aws.String(u.Bucket),
+		Key:         aws.String(p.Asset.S3Key),
+		Body:        bytes.NewReader(p.Content),
+		ContentType: aws.String(p.Asset.MIME),
+		Metadata:    p.Metadata,
+	}
+	if u.SSE != nil {
+		input.ServerSideEncryption = *u.SSE
+	}
+
+	_, err := u.Uploader.Upload(ctx, input)
+	return err
+}
+
 // UploadFileDirect uploads a file directly to S3 at the specified key (no deduplication)
 // This is used when you need to preserve the exact file structure
 func (u *S3Deps) UploadFileDirect(ctx context.Context, key string, content []byte, contentType string) (*model.Asset, error) {

src/server/api/go/internal/infra/blob/s3_test.go

@@ -0,0 +1,131 @@
+package blob
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"mime/multipart"
+	"net/textproto"
+	"strings"
+	"testing"
+
+	"github.com/bytedance/sonic"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPrepareJSONAsset(t *testing.T) {
+	s3 := &S3Deps{Bucket: "test-bucket"}
+
+	data := map[string]string{"hello": "world"}
+	prepared, err := s3.PrepareJSONAsset("parts/project-123", data)
+	require.NoError(t, err)
+
+	// Verify SHA256 matches manual computation
+	jsonBytes, _ := sonic.Marshal(data)
+	h := sha256.New()
+	h.Write(jsonBytes)
+	expectedSHA := hex.EncodeToString(h.Sum(nil))
+
+	assert.Equal(t, expectedSHA, prepared.Asset.SHA256)
+	assert.Equal(t, "test-bucket", prepared.Asset.Bucket)
+	assert.Equal(t, "application/json", prepared.Asset.MIME)
+	assert.Equal(t, int64(len(jsonBytes)), prepared.Asset.SizeB)
+	assert.Equal(t, jsonBytes, prepared.Content)
+
+	// Verify S3 key format: {prefix}/{date}/{sha256}.json
+	assert.True(t, strings.HasPrefix(prepared.Asset.S3Key, "parts/project-123/"))
+	assert.True(t, strings.HasSuffix(prepared.Asset.S3Key, expectedSHA+".json"))
+
+	// Verify metadata
+	assert.Equal(t, expectedSHA, prepared.Metadata["sha256"])
+}
+
+func TestPrepareJSONAsset_Deterministic(t *testing.T) {
+	s3 := &S3Deps{Bucket: "test-bucket"}
+
+	data := []int{1, 2, 3}
+	p1, err := s3.PrepareJSONAsset("prefix", data)
+	require.NoError(t, err)
+
+	p2, err := s3.PrepareJSONAsset("prefix", data)
+	require.NoError(t, err)
+
+	// Same input produces same SHA256 and key
+	assert.Equal(t, p1.Asset.SHA256, p2.Asset.SHA256)
+	assert.Equal(t, p1.Asset.S3Key, p2.Asset.S3Key)
+	assert.Equal(t, p1.Content, p2.Content)
+}
+
+func TestPrepareJSONAsset_DifferentData(t *testing.T) {
+	s3 := &S3Deps{Bucket: "test-bucket"}
+
+	p1, err := s3.PrepareJSONAsset("prefix", "aaa")
+	require.NoError(t, err)
+
+	p2, err := s3.PrepareJSONAsset("prefix", "bbb")
+	require.NoError(t, err)
+
+	assert.NotEqual(t, p1.Asset.SHA256, p2.Asset.SHA256)
+}
+
+// newTestFileHeader creates a multipart.FileHeader for testing.
+func newTestFileHeader(filename string, content []byte) *multipart.FileHeader {
+	var b bytes.Buffer
+	w := multipart.NewWriter(&b)
+	part, _ := w.CreatePart(textproto.MIMEHeader{
+		"Content-Disposition": {`form-data; name="file"; filename="` + filename + `"`},
+		"Content-Type":        {"application/octet-stream"},
+	})
+	part.Write(content)
+	w.Close()
+
+	r := multipart.NewReader(&b, w.Boundary())
+	form, _ := r.ReadForm(1 << 20)
+	return form.File["file"][0]
+}
+
+func TestPrepareFormFileAsset(t *testing.T) {
+	s3 := &S3Deps{Bucket: "test-bucket"}
+
+	content := []byte("test file content")
+	fh := newTestFileHeader("document.txt", content)
+
+	prepared, err := s3.PrepareFormFileAsset("assets/project-456", fh)
+	require.NoError(t, err)
+
+	// Verify SHA256
+	h := sha256.New()
+	h.Write(content)
+	expectedSHA := hex.EncodeToString(h.Sum(nil))
+
+	assert.Equal(t, expectedSHA, prepared.Asset.SHA256)
+	assert.Equal(t, "test-bucket", prepared.Asset.Bucket)
+	assert.Equal(t, int64(len(content)), prepared.Asset.SizeB)
+	assert.Equal(t, content, prepared.Content)
+
+	// Verify S3 key has correct extension
+	assert.True(t, strings.HasSuffix(prepared.Asset.S3Key, expectedSHA+".txt"))
+	assert.True(t, strings.HasPrefix(prepared.Asset.S3Key, "assets/project-456/"))
+
+	// Verify metadata includes filename
+	assert.Equal(t, expectedSHA, prepared.Metadata["sha256"])
+	assert.Equal(t, "document.txt", prepared.Metadata["name"])
+}
+
+func TestPrepareFormFileAsset_Deterministic(t *testing.T) {
+	s3 := &S3Deps{Bucket: "test-bucket"}
+
+	content := []byte("same content")
+	fh1 := newTestFileHeader("file.bin", content)
+	fh2 := newTestFileHeader("file.bin", content)
+
+	p1, err := s3.PrepareFormFileAsset("prefix", fh1)
+	require.NoError(t, err)
+
+	p2, err := s3.PrepareFormFileAsset("prefix", fh2)
+	require.NoError(t, err)
+
+	assert.Equal(t, p1.Asset.SHA256, p2.Asset.SHA256)
+	assert.Equal(t, p1.Asset.S3Key, p2.Asset.S3Key)
+}

src/server/api/go/internal/modules/service/session.go

@@ -293,6 +293,7 @@ func (s *sessionService) StoreMessage(ctx context.Context, in StoreMessageInput)
 
 	parts := make([]model.Part, 0, len(in.Parts))
 	var uploadedAssets []model.Asset
+	var pendingUploads []*blob.PreparedUpload
 
 	for idx := range in.Parts {
 		partIn := &in.Parts[idx] // Use pointer to avoid repeated indexing and allow modifications
@@ -316,14 +317,15 @@ func (s *sessionService) StoreMessage(ctx context.Context, in StoreMessageInput)
 				return nil, fmt.Errorf("parts[%d]: missing uploaded file %s", idx, partIn.FileField)
 			}
 
-			// upload asset to S3
-			asset, err := s.s3.UploadFormFile(ctx, "assets/"+in.ProjectID.String(), fh)
+			// Pre-compute asset metadata without S3 calls
+			prepared, err := s.s3.PrepareFormFileAsset("assets/"+in.ProjectID.String(), fh)
 			if err != nil {
-				return nil, fmt.Errorf("upload %s failed: %w", partIn.FileField, err)
+				return nil, fmt.Errorf("prepare %s failed: %w", partIn.FileField, err)
 			}
 
-			uploadedAssets = append(uploadedAssets, *asset)
-			part.Asset = asset
+			pendingUploads = append(pendingUploads, prepared)
+			uploadedAssets = append(uploadedAssets, prepared.Asset)
+			part.Asset = &prepared.Asset
 			part.Filename = fh.Filename
 		}
 
@@ -334,13 +336,37 @@ func (s *sessionService) StoreMessage(ctx context.Context, in StoreMessageInput)
 		parts = append(parts, part)
 	}
 
-	// upload parts to S3 as JSON file
-	asset, err := s.s3.UploadJSON(ctx, "parts/"+in.ProjectID.String(), parts)
+	// Pre-compute parts JSON asset metadata without S3 calls
+	partsAssetPrepared, err := s.s3.PrepareJSONAsset("parts/"+in.ProjectID.String(), parts)
 	if err != nil {
-		return nil, fmt.Errorf("upload parts to S3 failed: %w", err)
+		return nil, fmt.Errorf("prepare parts asset failed: %w", err)
 	}
 
-	uploadedAssets = append(uploadedAssets, *asset)
+	pendingUploads = append(pendingUploads, partsAssetPrepared)
+	partsAsset := partsAssetPrepared.Asset
+	uploadedAssets = append(uploadedAssets, partsAsset)
+
+	// Cache parts data in Redis before responding (uses pre-computed SHA256)
+	if s.redis != nil {
+		if err := s.cachePartsInRedis(ctx, partsAsset.SHA256, parts); err != nil {
+			s.log.Warn("failed to cache parts in Redis", zap.String("sha256", partsAsset.SHA256), zap.Error(err))
+		}
+	}
+
+	// Upload all assets to S3 asynchronously — not on the request critical path.
+	// Since S3 keys are content-addressed (SHA256), uploads are idempotent.
+	go func() {
+		bgCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer cancel()
+		for _, p := range pendingUploads {
+			if err := s.s3.UploadPrepared(bgCtx, p); err != nil {
+				s.log.Error("async S3 upload failed",
+					zap.String("s3_key", p.Asset.S3Key),
+					zap.String("sha256", p.Asset.SHA256),
+					zap.Error(err))
+			}
+		}
+	}()
 
 	// Increment asset reference counts asynchronously to avoid blocking the response.
 	go func() {
@@ -352,14 +378,6 @@ func (s *sessionService) StoreMessage(ctx context.Context, in StoreMessageInput)
 		}
 	}()
 
-	// Cache parts data in Redis after successful S3 upload
-	if s.redis != nil {
-		if err := s.cachePartsInRedis(ctx, asset.SHA256, parts); err != nil {
-			// Log error but don't fail the request if Redis caching fails
-			s.log.Warn("failed to cache parts in Redis", zap.String("sha256", asset.SHA256), zap.Error(err))
-		}
-	}
-
 	// Prepare message metadata
 	messageMeta := in.MessageMeta
 	if messageMeta == nil {
@@ -370,7 +388,7 @@ func (s *sessionService) StoreMessage(ctx context.Context, in StoreMessageInput)
 		SessionID:      in.SessionID,
 		Role:           in.Role,
 		Meta:           datatypes.NewJSONType(messageMeta), // Store message-level metadata
-		PartsAssetMeta: datatypes.NewJSONType(*asset),
+		PartsAssetMeta: datatypes.NewJSONType(partsAsset),
 		Parts:          parts,
 	}