weichen
9a545b1e69
Why: the user ruled out self-registration (2026-09-30): accounts are created only by admins who set an initial password; first login (and first login after any admin reset) must force a password change. Evidence: - migration v10 auth_accounts (users password/must_change/is_admin + sessions table; upgrade test walks v1<->v10 both directions). - argon2id hash/verify with PHC encoding; store tests cover hash roundtrip, duplicate-owner conflict, must-change lifecycle, session keep/except semantics. - e2e TestAdminAccountFlow: static login page reachable without auth, wrong password 401, login sets HttpOnly cookie + mustChangePassword, admin creates user (409 on duplicate), non-admin create 403, forced change clears flag and old password dies, admin reset revokes sessions and re-arms forced change, API-key channel untouched, 5-fail backoff 429, logout kills session. Full go test ./... green. Impact: middleware now accepts Bearer key OR session cookie and exempts static + /api/auth/login (also fixes the latent case where the login page itself could 401). Embedded no-auth mode unchanged. Refs: proposal 2026-09-30-ui-redesign-and-auth.md section 5 (admin-issued variant, user decision 2026-09-30)
109 lines
2.9 KiB
Go
109 lines
2.9 KiB
Go
// Package pgbus 是 PG 档的事件跨实例扇出（阶段 3）：
|
||
// 落盘事务提交后 NOTIFY agent_events '<conv>|<lastSeq>'；每实例一个 LISTEN 连接，
|
||
// 收到通知按 seq 从库取回事件并转发本地 bus（SSE 以 seq 去重，双投无害）。
|
||
// 多实例部署契约：run 在其归属实例内存中执行，会话流量需按会话亲和路由
|
||
// （网关/客户端择点）；本包只解决“其他实例的订阅者”可见性。
|
||
package pgbus
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"agent-system-backend/internal/bus"
|
||
"agent-system-backend/internal/store"
|
||
core "agent-system-core"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
const channel = "agent_events"
|
||
|
||
// Notifier 落盘后通知（任意连接池连接可用）。
|
||
type Notifier struct{ st *store.Store }
|
||
|
||
func NewNotifier(st *store.Store) *Notifier { return &Notifier{st: st} }
|
||
|
||
// Notify 在 AppendEvents 成功后调用：一批事件一条通知（载荷=会话|末 seq）。
|
||
func (n *Notifier) Notify(conversationID string, evs []core.StreamEvent) {
|
||
if len(evs) == 0 {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
defer cancel()
|
||
payload := conversationID + "|" + strconv.FormatInt(evs[len(evs)-1].Seq, 10)
|
||
if _, err := n.st.NotifyPG(ctx, channel, payload); err != nil {
|
||
log.Printf("[pgbus] notify %s: %v", conversationID, err)
|
||
}
|
||
}
|
||
|
||
// Listener 每实例一个：LISTEN → 按 seq 取回 → 本地 bus 转发。
|
||
type Listener struct {
|
||
conn *pgx.Conn
|
||
st *store.Store
|
||
bus *bus.Bus
|
||
done chan struct{}
|
||
}
|
||
|
||
// Start 建立监听连接并开始转发循环；返回后即可用，随 ctx 取消停机。
|
||
func Start(ctx context.Context, dsn string, st *store.Store, b *bus.Bus) (*Listener, error) {
|
||
conn, err := pgx.Connect(ctx, dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := conn.Exec(ctx, "LISTEN "+channel); err != nil {
|
||
conn.Close(ctx)
|
||
return nil, err
|
||
}
|
||
l := &Listener{conn: conn, st: st, bus: b, done: make(chan struct{})}
|
||
go l.loop(ctx)
|
||
return l, nil
|
||
}
|
||
|
||
func (l *Listener) loop(ctx context.Context) {
|
||
defer close(l.done)
|
||
for {
|
||
n, err := l.conn.WaitForNotification(ctx)
|
||
if err != nil {
|
||
if ctx.Err() != nil {
|
||
return // 停机
|
||
}
|
||
log.Printf("[pgbus] wait: %v（1s 后重连）", err)
|
||
time.Sleep(time.Second)
|
||
continue
|
||
}
|
||
conv, seq, ok := parsePayload(n.Payload)
|
||
if !ok {
|
||
continue
|
||
}
|
||
ev, err := l.st.EventAt(conv, seq)
|
||
if err != nil {
|
||
continue // 事件不可见（极早期窗口）——SSE 重放兜底
|
||
}
|
||
l.bus.Publish(ev)
|
||
}
|
||
}
|
||
|
||
func parsePayload(p string) (string, int64, bool) {
|
||
i := strings.LastIndex(p, "|")
|
||
if i < 0 {
|
||
return "", 0, false
|
||
}
|
||
seq, err := strconv.ParseInt(p[i+1:], 10, 64)
|
||
if err != nil {
|
||
return "", 0, false
|
||
}
|
||
return p[:i], seq, true
|
||
}
|
||
|
||
// Close 关闭监听连接。
|
||
func (l *Listener) Close(ctx context.Context) {
|
||
l.conn.Close(ctx)
|
||
select {
|
||
case <-l.done:
|
||
case <-time.After(2 * time.Second):
|
||
}
|
||
}
|