e1326acaf9
- 3.1 bcrypt 移出事务 (Register): repository.HashPassword 前移到 db.Transaction 之前,消除连接池占用。
- 3.2 Login 消除用户枚举 + 限流 + timing 抹平: pkg/errors 加 ErrInvalidCredential
/ErrTooManyLoginAttempts; 用户不存在/密码错/密码空 三路径统一返回同一错误;
mobile 5次/ip 20次 per 15min 限流 (Redis, fail-open 降级); user-not-found 走
dummy bcrypt 抹平 ~100ms 时序差,完全消除枚举侧信道;空密码分支已核实无时序 leak。
- 3.3 MQ streams adapter 停用 → stub: 0 业务调用方, 新 stub EventProducer.Publish no-op;
pkg/mq/mq.go Init 不再装配 streams; 全仓 grep 验证 11 处硬编码
'gallery'/'default' 集中到 pkg/queue/consts (值不变, 仅消漂移)。
- 3.5 JWT 密钥治理: pkg/jwt MustInit fail-fast + atomic.Value (见上一个 commit 293c7b1)。
- 3.6 aiChat 健壮性: SaveContext 用 persona.ID(非 req.PersonaId); Redis/memory 错误
记 WARN 不静默; Dify err 映射稳定用户文案,原始 err 仅服务端日志。
- 3.7 statistic.Client 重构: TrackEvent 改 buffered channel (cap 1024) + dispatchLoop
worker; 失败 ERROR 日志带字段; drop 记 WARN; Close 可重复调用。
- 3.8 网关聚合: StarCache (60s TTL, singleflight) 替换 5+ 处 GetFanIdentities 链式调用;
DeleteAccount 改网关直调 userService.DeleteAccount(避免改 hand-written triple.go
风险,见报告 §5 proto 风险复盘); 铸造双写改异步 channel+consumer (3 retry)。
- 大量单测: 各子项 TDD (RED→GREEN), 关键并发 race_test (50 goroutine)。
Co-Authored-By: Claude <noreply@anthropic.com>
79 lines
2.5 KiB
Go
79 lines
2.5 KiB
Go
package streams
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/topfans/backend/pkg/logger"
|
|
"github.com/topfans/backend/pkg/mq/adapter"
|
|
)
|
|
|
|
// Config is retained for source compatibility while the Redis Streams adapter
|
|
// is disabled.
|
|
type Config struct {
|
|
RedisAddr string
|
|
RedisDB int
|
|
Password string
|
|
ConsumerGroup string
|
|
MaxLen int64
|
|
ReadBlockMS int
|
|
}
|
|
|
|
var publishWarnOnce sync.Once
|
|
|
|
type stubProducer struct{}
|
|
|
|
func (stubProducer) Publish(_ context.Context, _ string, _ adapter.Event) error {
|
|
publishWarnOnce.Do(func() {
|
|
if logger.Logger != nil {
|
|
logger.Logger.Warn("streams.Publish called but streams adapter is deprecated; events should go via pkg/statistic.TrackEvent")
|
|
}
|
|
})
|
|
return nil
|
|
}
|
|
|
|
type stubConsumer struct{}
|
|
|
|
func (stubConsumer) Subscribe(_ context.Context, _ []string, _ string, _ adapter.EventHandler) error {
|
|
return nil
|
|
}
|
|
|
|
func (stubConsumer) Ack(_ context.Context, _, _, _ string) error { return nil }
|
|
func (stubConsumer) Run(_ context.Context) error { return nil }
|
|
func (stubConsumer) Stop() error { return nil }
|
|
|
|
// NewStubProducer returns the process-wide event producer compatibility stub.
|
|
func NewStubProducer() adapter.EventProducer { return stubProducer{} }
|
|
|
|
// NewStubConsumer returns the process-wide event consumer compatibility stub.
|
|
func NewStubConsumer() adapter.EventConsumer { return stubConsumer{} }
|
|
|
|
// Adapter retains the former streams adapter's exported API as a no-op stub.
|
|
type Adapter struct{}
|
|
|
|
// New retains constructor compatibility without creating a Redis client.
|
|
func New(_ Config) (*Adapter, error) { return &Adapter{}, nil }
|
|
|
|
func (a *Adapter) Publish(ctx context.Context, topic string, event adapter.Event) error {
|
|
return stubProducer{}.Publish(ctx, topic, event)
|
|
}
|
|
|
|
func (a *Adapter) Subscribe(ctx context.Context, topics []string, group string, handler adapter.EventHandler) error {
|
|
return stubConsumer{}.Subscribe(ctx, topics, group, handler)
|
|
}
|
|
|
|
func (a *Adapter) Ack(ctx context.Context, topic, group, msgID string) error {
|
|
return stubConsumer{}.Ack(ctx, topic, group, msgID)
|
|
}
|
|
|
|
func (a *Adapter) Run(ctx context.Context) error { return stubConsumer{}.Run(ctx) }
|
|
func (a *Adapter) Stop() error { return stubConsumer{}.Stop() }
|
|
func (a *Adapter) Close() error { return nil }
|
|
|
|
var (
|
|
_ adapter.EventProducer = (*Adapter)(nil)
|
|
_ adapter.EventConsumer = (*Adapter)(nil)
|
|
_ adapter.EventProducer = stubProducer{}
|
|
_ adapter.EventConsumer = stubConsumer{}
|
|
)
|