perf(api): buffer asset_references writes via Redis to eliminate row-lock contention (#467)

Under high concurrency (50+ concurrent learn sessions), BatchIncrementAssetRefs
causes PostgreSQL row-level lock contention on the asset_references table,
resulting in slow queries (avg 316ms), S3 context canceled errors, and 504s.

This replaces the synchronous DB upsert in StoreMessage with an async
Redis-buffered writer that:
- Uses HINCRBY to atomically accumulate ref counts in Redis (µs latency)
- Flushes coalesced batches to DB every 1s via a background goroutine
- Falls back to synchronous DB write if Redis enqueue fails
- Restores counts to Redis on DB flush failure for automatic retry
- Drains remaining data on graceful shutdown

The writer is stateless — any pod can flush any project's pending data,
and pod crash/restart loses at most 1s of ref_count increments (which are
only used for storage analytics and GC, not hard constraints).

Author: GenerQAQ

src/server/api/go/cmd/server/main.go

@@ -24,6 +24,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/memodb-io/Acontext/internal/bootstrap"
 	"github.com/memodb-io/Acontext/internal/config"
+	"github.com/memodb-io/Acontext/internal/infra/assetrefwriter"
 	"github.com/memodb-io/Acontext/internal/infra/cache"
 	dbpkg "github.com/memodb-io/Acontext/internal/infra/db"
 	"github.com/memodb-io/Acontext/internal/modules/handler"
@@ -138,5 +139,13 @@ func main() {
 	if err := srv.Shutdown(ctx); err != nil {
 		log.Sugar().Errorw("server shutdown", "err", err)
 	}
+
+	// Flush buffered asset reference writes before exit
+	if writer := do.MustInvoke[*assetrefwriter.AssetRefWriter](inj); writer != nil {
+		if err := writer.Close(ctx); err != nil {
+			log.Sugar().Errorw("asset ref writer shutdown", "err", err)
+		}
+	}
+
 	log.Sugar().Info("server exited")
 }

src/server/api/go/internal/bootstrap/container.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/memodb-io/Acontext/configs"
 	"github.com/memodb-io/Acontext/internal/config"
+	"github.com/memodb-io/Acontext/internal/infra/assetrefwriter"
 	"github.com/memodb-io/Acontext/internal/infra/blob"
 	"github.com/memodb-io/Acontext/internal/infra/cache"
 	"github.com/memodb-io/Acontext/internal/infra/db"
@@ -215,6 +216,27 @@ func BuildContainer() *do.Injector {
 			do.MustInvoke[*blob.S3Deps](i),
 		), nil
 	})
+
+	// AssetRefWriter — buffers asset reference increments in Redis for async batch flush
+	do.Provide(inj, func(i *do.Injector) (*assetrefwriter.AssetRefWriter, error) {
+		cfg := do.MustInvoke[*config.Config](i)
+		if !cfg.AssetRefWriter.Enabled {
+			return nil, nil
+		}
+		interval := time.Duration(cfg.AssetRefWriter.FlushIntervalMs) * time.Millisecond
+		if interval <= 0 {
+			interval = time.Second
+		}
+		w := assetrefwriter.New(
+			do.MustInvoke[*redis.Client](i),
+			do.MustInvoke[repo.AssetReferenceRepo](i),
+			do.MustInvoke[*zap.Logger](i),
+			assetrefwriter.WithFlushInterval(interval),
+		)
+		w.Start()
+		return w, nil
+	})
+
 	do.Provide(inj, func(i *do.Injector) (repo.SessionRepo, error) {
 		return repo.NewSessionRepo(
 			do.MustInvoke[*gorm.DB](i),
@@ -268,6 +290,7 @@ func BuildContainer() *do.Injector {
 			do.MustInvoke[repo.SessionRepo](i),
 			do.MustInvoke[repo.SessionEventRepo](i),
 			do.MustInvoke[repo.AssetReferenceRepo](i),
+			do.MustInvoke[*assetrefwriter.AssetRefWriter](i),
 			do.MustInvoke[*zap.Logger](i),
 			do.MustInvoke[*blob.S3Deps](i),
 			do.MustInvoke[*mq.Publisher](i),

src/server/api/go/internal/config/config.go

@@ -101,19 +101,25 @@ type ArtifactCfg struct {
 	MaxUploadSizeBytes int64 // Maximum file upload size in bytes
 }
 
+type AssetRefWriterCfg struct {
+	Enabled         bool // Enable async buffered writes for asset references (default true)
+	FlushIntervalMs int  // Flush interval in milliseconds (default 1000)
+}
+
 type Config struct {
-	App       AppCfg
-	Root      RootCfg
-	Log       LogCfg
-	Database  DBCfg
-	Redis     RedisCfg
-	RabbitMQ  MQCfg
-	S3        S3Cfg
-	Core      CoreCfg
-	Metrics   MetricsCfg
-	Telemetry TelemetryCfg
-	Supabase  SupabaseCfg
-	Artifact  ArtifactCfg
+	App            AppCfg
+	Root           RootCfg
+	Log            LogCfg
+	Database       DBCfg
+	Redis          RedisCfg
+	RabbitMQ       MQCfg
+	S3             S3Cfg
+	Core           CoreCfg
+	Metrics        MetricsCfg
+	Telemetry      TelemetryCfg
+	Supabase       SupabaseCfg
+	Artifact       ArtifactCfg
+	AssetRefWriter AssetRefWriterCfg
 }
 
 func setDefaults(v *viper.Viper) {
@@ -151,6 +157,8 @@ func setDefaults(v *viper.Viper) {
 	v.SetDefault("supabase.apiKey", "")
 	v.SetDefault("supabase.authURL", "")
 	v.SetDefault("artifact.maxUploadSizeBytes", 16777216) // Default 16MB (16 * 1024 * 1024 bytes)
+	v.SetDefault("assetRefWriter.enabled", true)
+	v.SetDefault("assetRefWriter.flushIntervalMs", 1000)
 }
 
 func Load() (*Config, error) {