Files
LLMGuardX Dev ea78ef5674 0.11.8: 优化方向落地(流式聊天/操作审计/列表分页/测试补齐)
P1-聊天 SSE 流式响应:
- 新增 POST /chat/sessions/{id}/messages/stream:网关 text/event-stream 实时
  透传,流结束整轮落库(哈希链),上游忽略 stream 返回普通 JSON 时自动转
  SSE 事件,非 2xx 错误缓冲后走统一错误处理(不落 header);
- 前端 fetch+ReadableStream 解析 SSE,占位气泡实时填充,支持停止生成
  (AbortController),切会话丢弃迟到增量防串扰。

P2-管理操作审计(admin_op_logs):
- 新表+oplog 包(同步写,失败不阻塞业务);管理端查询端点
  GET /api/v1/admin/op-logs(操作者/类型过滤+分页,audit:read);
- 埋点:渠道 save/delete/grant(幂等重复不重复记)/revoke_grant、账号
  create/update、角色 CRUD、API Key create/revoke/limits、工具
  save/delete、审批决定(资源/工具)、模型配额;管理端「操作审计」菜单。

P2-列表分页与安全上限:
- 用户/管理员列表 q+limit+offset 分页(默认 50 上限 200),渠道授权弹窗
  改远程搜索,不再全量拉取 portal-users;api_keys/channels List 加
  LIMIT 200 防全表扫描。

健壮性:
- ChatModels/approvedModel 对 decided_at 为 NULL 的历史批准记录
  COALESCE 兜底,修复 NULL scan 报错;
- docker-compose 补 ALLOW_PRIVATE_PROVIDER_URLS 透传(默认 false)。

测试:
- portal: 流式解析/错误提取/stream writer 模式单测,会话生命周期/哈希链
  完整性/200 条上限/busy 租约回收集成测试;
- channel: CRUD+加解密+部门可见性+授权撤销+幂等+审计落库集成测试。
  全部通过;全量 go vet 干净。
2026-08-13 16:15:02 +08:00

633 lines
23 KiB
Go

package identity
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"regexp"
"sort"
"strconv"
"strings"
"aigateway.local/core/internal/platform/apiresponse"
platformid "aigateway.local/core/internal/platform/id"
"aigateway.local/core/internal/platform/oplog"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
)
var (
ErrIdentityConflict = errors.New("identity already exists")
permissionPattern = regexp.MustCompile(`^[a-z][a-z0-9_.:-]{2,127}$`)
)
type ManagementHTTPHandler struct {
service *Service
mux *http.ServeMux
// license 提供账号数上限管控(nil 时不限制)。
license interface {
AccountLimit() int
}
}
// SetLicenseManager 注入 License 管理器用于账号数管控。
func (h *ManagementHTTPHandler) SetLicenseManager(manager interface{ AccountLimit() int }) {
h.license = manager
}
// checkAccountLimit 在创建账号前校验 License 账号数上限。
func (h *ManagementHTTPHandler) checkAccountLimit(writer http.ResponseWriter, request *http.Request) bool {
if h.license == nil {
return true
}
limit := h.license.AccountLimit()
if limit <= 0 {
return true // 不限
}
var total int
err := h.service.repository.CountIdentities(request.Context(), &total)
if err != nil {
apiresponse.Error(writer, http.StatusServiceUnavailable, "身份服务暂不可用")
return false
}
if total >= limit {
apiresponse.Error(writer, http.StatusForbidden, fmt.Sprintf("账号数已达 License 上限(%d 个),请联系管理员升级", limit))
return false
}
return true
}
type identityInput struct {
Login string `json:"login"`
DisplayName string `json:"display_name"`
Role string `json:"role"`
Password *string `json:"password"`
Permissions []string `json:"permissions"`
Active *bool `json:"active"`
DepartmentID *string `json:"department_id"`
}
func NewManagementHTTPHandler(service *Service) *ManagementHTTPHandler {
h := &ManagementHTTPHandler{service: service, mux: http.NewServeMux()}
h.mux.HandleFunc("GET /api/v1/admin/identities/admins", h.list(KindAdmin))
h.mux.HandleFunc("POST /api/v1/admin/identities/admins", h.create(KindAdmin))
h.mux.HandleFunc("PUT /api/v1/admin/identities/admins/{identity_id}", h.update(KindAdmin))
h.mux.HandleFunc("GET /api/v1/admin/identities/portal-users", h.list(KindPortal))
h.mux.HandleFunc("GET /api/v1/admin/roles", h.listRoles)
h.mux.HandleFunc("POST /api/v1/admin/roles", h.createRole)
h.mux.HandleFunc("PUT /api/v1/admin/roles/{role_id}", h.updateRole)
h.mux.HandleFunc("DELETE /api/v1/admin/roles/{role_id}", h.deleteRole)
h.mux.HandleFunc("POST /api/v1/admin/identities/portal-users", h.create(KindPortal))
h.mux.HandleFunc("PUT /api/v1/admin/identities/portal-users/{identity_id}", h.update(KindPortal))
h.mux.HandleFunc("GET /api/v1/admin/departments", h.listDepartments)
h.mux.HandleFunc("POST /api/v1/admin/departments", h.createDepartment)
h.mux.HandleFunc("PUT /api/v1/admin/departments/{department_id}", h.updateDepartment)
h.registerOIDC()
h.registerSocialAdmin()
return h
}
func (h *ManagementHTTPHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
h.mux.ServeHTTP(writer, request)
}
func (h *ManagementHTTPHandler) list(kind Kind) http.HandlerFunc {
return func(writer http.ResponseWriter, request *http.Request) {
if _, ok := h.requirePermission(writer, request); !ok {
return
}
// 分页 + 关键字过滤:企业用户量级可能很大,不允许全量返回。
// 默认每页 50,上限 200;q 匹配账号/显示名称前缀(不区分大小写)。
limit, err := strconv.Atoi(request.URL.Query().Get("limit"))
if err != nil || limit < 1 || limit > 200 {
limit = 50
}
offset, _ := strconv.Atoi(request.URL.Query().Get("offset"))
if offset < 0 {
offset = 0
}
q := strings.TrimSpace(request.URL.Query().Get("q"))
accounts, total, err := h.service.repository.ListIdentities(request.Context(), kind, q, limit, offset)
if err != nil {
h.writeError(writer, err)
return
}
items := make([]map[string]any, 0, len(accounts))
for _, account := range accounts {
items = append(items, managementView(account))
}
apiresponse.OK(writer, map[string]any{"items": items, "total": total, "limit": limit, "offset": offset})
}
}
func (h *ManagementHTTPHandler) create(kind Kind) http.HandlerFunc {
return func(writer http.ResponseWriter, request *http.Request) {
actor, ok := h.requirePermission(writer, request)
if !ok {
return
}
if !h.checkAccountLimit(writer, request) {
return
}
input, account, password, ok := h.decode(writer, request, kind, true)
if !ok {
return
}
_ = input
passwordHash, err := h.service.hasher.Hash(password)
if err != nil {
h.writeError(writer, err)
return
}
account.PasswordHash = passwordHash
created, err := h.service.repository.CreateIdentity(request.Context(), account, actor.ID)
if err != nil {
h.writeError(writer, err)
return
}
apiresponse.OK(writer, managementView(created))
}
}
func (h *ManagementHTTPHandler) update(kind Kind) http.HandlerFunc {
return func(writer http.ResponseWriter, request *http.Request) {
actor, ok := h.requirePermission(writer, request)
if !ok {
return
}
input, account, password, ok := h.decode(writer, request, kind, false)
if !ok {
return
}
account.ID = request.PathValue("identity_id")
current, err := h.service.findByID(request.Context(), kind, account.ID)
if err != nil {
h.writeError(writer, err)
return
}
// 部分更新语义:未提供的字段保留当前值。否则"只改显示名"的 PUT 会
// 把角色重置为默认 operator、把停用账号重新激活,造成意外的权限变更。
if account.Role == "" {
account.Role = current.Role
}
if input.Active == nil {
account.Active = current.Active
}
if kind == KindAdmin && actor.ID == current.ID && (account.Role != current.Role || !account.Active) {
apiresponse.Error(writer, http.StatusConflict, "不能停用自身账号或修改自身角色")
return
}
var passwordHash *string
if password != "" {
hash, err := h.service.hasher.Hash(password)
if err != nil {
h.writeError(writer, err)
return
}
passwordHash = &hash
}
updated, err := h.service.repository.UpdateIdentity(request.Context(), account, passwordHash, actor.ID)
if err != nil {
h.writeError(writer, err)
return
}
if password != "" {
// 管理员重置密码属于凭据变更:立即作废该账号的全部既有会话,
// 与自助改密/2FA 变更的语义保持一致。
h.service.sessions.BumpAuthVersion(request.Context(), kind, account.ID)
}
apiresponse.OK(writer, managementView(updated))
}
}
func (h *ManagementHTTPHandler) listRoles(writer http.ResponseWriter, request *http.Request) {
actor, ok := h.requirePermission(writer, request)
if !ok {
return
}
_ = actor
roles, err := h.service.repository.ListRoles(request.Context())
if err != nil {
h.writeError(writer, err)
return
}
apiresponse.OK(writer, roles)
}
type roleInput struct {
Code string `json:"code"`
Name string `json:"name"`
Description string `json:"description"`
Permissions []string `json:"permissions"`
}
func (h *ManagementHTTPHandler) decodeRole(writer http.ResponseWriter, request *http.Request) (roleInput, bool) {
var input roleInput
decoder := json.NewDecoder(http.MaxBytesReader(writer, request.Body, 1<<20))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&input); err != nil {
apiresponse.Error(writer, http.StatusBadRequest, "请求格式无效")
return input, false
}
input.Code = strings.ToLower(strings.TrimSpace(input.Code))
input.Name = strings.TrimSpace(input.Name)
input.Description = strings.TrimSpace(input.Description)
if !roleCodePattern.MatchString(input.Code) || input.Name == "" || len(input.Name) > 64 || len(input.Description) > 512 {
apiresponse.Error(writer, http.StatusBadRequest, "角色代码或名称格式无效")
return input, false
}
permissions, err := normalizePermissions(input.Permissions)
if err != nil {
apiresponse.Error(writer, http.StatusBadRequest, err.Error())
return input, false
}
input.Permissions = permissions
return input, true
}
func (h *ManagementHTTPHandler) createRole(writer http.ResponseWriter, request *http.Request) {
actor, ok := h.requirePermission(writer, request)
if !ok {
return
}
input, ok := h.decodeRole(writer, request)
if !ok {
return
}
role, err := h.service.repository.SaveRole(request.Context(), "", input.Code, input.Name, input.Description, input.Permissions, actor.ID, true)
if err != nil {
h.writeError(writer, err)
return
}
apiresponse.OK(writer, role)
}
func (h *ManagementHTTPHandler) updateRole(writer http.ResponseWriter, request *http.Request) {
actor, ok := h.requirePermission(writer, request)
if !ok {
return
}
input, ok := h.decodeRole(writer, request)
if !ok {
return
}
role, err := h.service.repository.SaveRole(request.Context(), request.PathValue("role_id"), input.Code, input.Name, input.Description, input.Permissions, actor.ID, false)
if err != nil {
h.writeError(writer, err)
return
}
apiresponse.OK(writer, role)
}
func (h *ManagementHTTPHandler) deleteRole(writer http.ResponseWriter, request *http.Request) {
actor, ok := h.requirePermission(writer, request)
if !ok {
return
}
if err := h.service.repository.DeleteRole(request.Context(), request.PathValue("role_id"), actor.ID); err != nil {
h.writeError(writer, err)
return
}
apiresponse.OK(writer, map[string]bool{"deleted": true})
}
func (h *ManagementHTTPHandler) decode(writer http.ResponseWriter, request *http.Request, kind Kind, creating bool) (identityInput, Account, string, bool) {
var input identityInput
decoder := json.NewDecoder(http.MaxBytesReader(writer, request.Body, 1<<20))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&input); err != nil {
apiresponse.Error(writer, http.StatusBadRequest, "请求格式无效")
return input, Account{}, "", false
}
input.Login = strings.ToLower(strings.TrimSpace(input.Login))
input.DisplayName = strings.TrimSpace(input.DisplayName)
input.Role = strings.ToLower(strings.TrimSpace(input.Role))
if input.Role == "" {
// 创建时缺省角色;更新时留空表示"不修改该字段",
// 由 update() 保留当前值,避免只改显示名就静默重置角色。
if creating {
if kind == KindPortal {
input.Role = "member"
} else {
input.Role = "operator"
}
}
}
if len(input.Login) < 2 || len(input.Login) > 128 || len(input.DisplayName) > 64 {
apiresponse.Error(writer, http.StatusBadRequest, "账号或显示名称格式无效")
return input, Account{}, "", false
}
if kind == KindAdmin && input.Role != "" {
switch input.Role {
case "superadmin", "operator", "auditor":
default:
// 自定义角色:必须存在于角色表,并把其权限展开到账号
// permissions(角色权限变更后由管理员重新分配或手动同步)。
role, roleErr := h.service.repository.FindRole(request.Context(), input.Role)
if roleErr != nil {
apiresponse.Error(writer, http.StatusBadRequest, "角色不存在")
return input, Account{}, "", false
}
input.Permissions = append(input.Permissions, role.Permissions...)
}
}
if kind == KindPortal && input.Role != "" && input.Role != "member" {
apiresponse.Error(writer, http.StatusBadRequest, "门户角色无效")
return input, Account{}, "", false
}
permissions, err := normalizePermissions(input.Permissions)
if err != nil {
apiresponse.Error(writer, http.StatusBadRequest, err.Error())
return input, Account{}, "", false
}
password := ""
if input.Password != nil {
password = *input.Password
}
if creating && len(password) < 12 || password != "" && len(password) < 12 || len(password) > 1024 {
apiresponse.Error(writer, http.StatusBadRequest, "口令长度必须为 12 至 1024 个字符")
return input, Account{}, "", false
}
active := true
if input.Active != nil {
active = *input.Active
}
var departmentID *string
if input.DepartmentID != nil && strings.TrimSpace(*input.DepartmentID) != "" {
value := strings.TrimSpace(*input.DepartmentID)
department, err := h.service.repository.GetDepartment(request.Context(), value)
if err != nil || !department.Active {
apiresponse.Error(writer, http.StatusBadRequest, "所选部门不存在或已停用")
return input, Account{}, "", false
}
departmentID = &value
}
return input, Account{
Kind: kind, Login: input.Login, DisplayName: input.DisplayName,
Role: input.Role, Permissions: permissions, Active: active, DepartmentID: departmentID,
}, password, true
}
func (h *ManagementHTTPHandler) requirePermission(writer http.ResponseWriter, request *http.Request) (Account, bool) {
account, err := h.service.Authenticate(request.Context(), KindAdmin, request.Header.Get("Authorization"))
if err != nil {
apiresponse.Error(writer, http.StatusUnauthorized, "登录状态无效或已过期")
return Account{}, false
}
if !HasPermission(account, PermissionIdentityManage) {
apiresponse.Error(writer, http.StatusForbidden, "缺少身份管理权限")
return Account{}, false
}
return account, true
}
var roleCodePattern = regexp.MustCompile(`^[a-z][a-z0-9_-]{1,63}$`)
func (h *ManagementHTTPHandler) writeError(writer http.ResponseWriter, err error) {
switch {
case errors.Is(err, ErrNotFound):
apiresponse.Error(writer, http.StatusNotFound, "账号不存在")
case errors.Is(err, ErrIdentityConflict):
apiresponse.Error(writer, http.StatusConflict, "账号已存在")
case errors.Is(err, ErrUnavailable):
apiresponse.Error(writer, http.StatusServiceUnavailable, "身份管理服务暂不可用")
default:
apiresponse.Error(writer, http.StatusInternalServerError, "身份管理操作失败")
}
}
func normalizePermissions(values []string) ([]string, error) {
seen := make(map[string]struct{})
result := make([]string, 0, len(values))
for _, value := range values {
value = strings.ToLower(strings.TrimSpace(value))
if value != "*" && !permissionPattern.MatchString(value) {
return nil, fmt.Errorf("权限字符串 %q 格式无效", value)
}
if _, exists := seen[value]; exists {
continue
}
seen[value] = struct{}{}
result = append(result, value)
}
sort.Strings(result)
return result, nil
}
func managementView(account Account) map[string]any {
return map[string]any{
"id": account.ID, "kind": account.Kind, "login": account.Login,
"display_name": account.DisplayName, "role": account.Role,
"permissions": account.Permissions, "effective_permissions": EffectivePermissions(account),
"active": account.Active, "auth_source": account.AuthSource,
"totp_enabled": account.TOTPEnabled, "locked_until": account.LockedUntil,
"created_at": account.CreatedAt, "updated_at": account.UpdatedAt,
"department_id": account.DepartmentID, "department_name": account.DepartmentName,
}
}
func (r *Repository) ListIdentities(ctx context.Context, kind Kind, q string, limit, offset int) ([]Account, int, error) {
if r.pool == nil {
return nil, 0, ErrUnavailable
}
// 关键字同时匹配账号与显示名称(前缀,不区分大小写)。
filter := ""
args := []any{}
if q != "" {
filter = ` AND (lower({{login}}) LIKE $1 OR lower({{name}}) LIKE $1)`
args = append(args, strings.ToLower(q)+"%")
}
var query string
if kind == KindPortal {
query = `
SELECT u.id::text, u.account, u.name, u.role, u.permissions, u.active,
u.totp_enabled, u.locked_until, u.auth_source, u.created_at, u.updated_at,
u.department_id::text, COALESCE(d.name, '')
FROM gateway.portal_users u
LEFT JOIN gateway.departments d ON d.id = u.department_id
WHERE 1=1` + filter + `
ORDER BY lower(u.account)
LIMIT $` + fmt.Sprintf("%d", len(args)+1) + ` OFFSET $` + fmt.Sprintf("%d", len(args)+2)
query = strings.ReplaceAll(query, "{{login}}", "u.account")
query = strings.ReplaceAll(query, "{{name}}", "u.name")
} else {
query = `
SELECT id::text, username, display_name, role, permissions, active,
totp_enabled, locked_until, 'local', created_at, updated_at
FROM gateway.admin_accounts
WHERE 1=1` + filter + `
ORDER BY lower(username)
LIMIT $` + fmt.Sprintf("%d", len(args)+1) + ` OFFSET $` + fmt.Sprintf("%d", len(args)+2)
query = strings.ReplaceAll(query, "{{login}}", "username")
query = strings.ReplaceAll(query, "{{name}}", "display_name")
}
args = append(args, limit, offset)
rows, err := r.pool.Query(ctx, query, args...)
if err != nil {
return nil, 0, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
defer rows.Close()
accounts := make([]Account, 0)
for rows.Next() {
account := Account{Kind: kind}
arguments := []any{
&account.ID, &account.Login, &account.DisplayName, &account.Role,
&account.Permissions, &account.Active, &account.TOTPEnabled,
&account.LockedUntil, &account.AuthSource, &account.CreatedAt, &account.UpdatedAt,
}
if kind == KindPortal {
arguments = append(arguments, &account.DepartmentID, &account.DepartmentName)
}
if err := rows.Scan(arguments...); err != nil {
return nil, 0, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
accounts = append(accounts, account)
}
if err := mapRepositoryError(rows.Err()); err != nil {
return nil, 0, err
}
// 总数:同条件 count,用于前端分页。
var total int
countFilter := strings.ReplaceAll(filter, "{{login}}", "u.account")
countFilter = strings.ReplaceAll(countFilter, "{{name}}", "u.name")
countTable := "gateway.portal_users u"
if kind != KindPortal {
countFilter = strings.ReplaceAll(filter, "{{login}}", "username")
countFilter = strings.ReplaceAll(countFilter, "{{name}}", "display_name")
countTable = "gateway.admin_accounts"
}
// args 末尾两个是 limit/offset,count 只用前面的过滤参数。
countArgs := args[:len(args)-2]
if err := r.pool.QueryRow(ctx, `SELECT count(*) FROM `+countTable+` WHERE 1=1`+countFilter, countArgs...).Scan(&total); err != nil {
return nil, 0, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
return accounts, total, nil
}
func (r *Repository) CreateIdentity(ctx context.Context, account Account, actorID string) (Account, error) {
if r.pool == nil {
return Account{}, ErrUnavailable
}
id, err := platformid.NewUUID()
if err != nil {
return Account{}, err
}
eventID, err := platformid.NewUUID()
if err != nil {
return Account{}, err
}
account.ID = id
tx, err := r.pool.Begin(ctx)
if err != nil {
return Account{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
defer func() { _ = tx.Rollback(ctx) }()
if account.Kind == KindAdmin {
err = tx.QueryRow(ctx, `
INSERT INTO gateway.admin_accounts
(id, username, display_name, role, permissions, password_hash, active)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING created_at, updated_at`, account.ID, account.Login, account.DisplayName,
account.Role, account.Permissions, account.PasswordHash, account.Active,
).Scan(&account.CreatedAt, &account.UpdatedAt)
} else {
account.AuthSource = "local"
err = tx.QueryRow(ctx, `
INSERT INTO gateway.portal_users
(id, account, name, role, permissions, password_hash, auth_source, active, department_id)
VALUES ($1, $2, $3, $4, $5, $6, 'local', $7, $8)
RETURNING created_at, updated_at`, account.ID, account.Login, account.DisplayName,
account.Role, account.Permissions, account.PasswordHash, account.Active, account.DepartmentID,
).Scan(&account.CreatedAt, &account.UpdatedAt)
}
if err != nil {
return Account{}, mapManagementError(err)
}
payload, _ := json.Marshal(map[string]any{"identity_id": account.ID, "kind": account.Kind, "actor_id": actorID})
if _, err := tx.Exec(ctx, `
INSERT INTO gateway.outbox_events
(event_id, event_type, event_version, aggregate_type, aggregate_id, payload)
VALUES ($1, 'identity.created', 1, 'identity', $2, $3)`, eventID, account.ID, payload); err != nil {
return Account{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
if err := tx.Commit(ctx); err != nil {
return Account{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
oplog.Record(ctx, r.pool, nil, actorID, account.Login, "identity.create", "identity", account.ID, map[string]any{
"kind": account.Kind, "role": account.Role, "active": account.Active,
"department_id": account.DepartmentID,
})
return account, nil
}
func (r *Repository) UpdateIdentity(ctx context.Context, account Account, passwordHash *string, actorID string) (Account, error) {
if r.pool == nil {
return Account{}, ErrUnavailable
}
eventID, err := platformid.NewUUID()
if err != nil {
return Account{}, err
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return Account{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
defer func() { _ = tx.Rollback(ctx) }()
if account.Kind == KindAdmin {
err = tx.QueryRow(ctx, `
UPDATE gateway.admin_accounts
SET username = $2, display_name = $3, role = $4, permissions = $5,
active = $6, password_hash = COALESCE($7, password_hash),
updated_at = clock_timestamp()
WHERE id = $1
RETURNING totp_enabled, locked_until, created_at, updated_at`, account.ID, account.Login,
account.DisplayName, account.Role, account.Permissions, account.Active, passwordHash,
).Scan(&account.TOTPEnabled, &account.LockedUntil, &account.CreatedAt, &account.UpdatedAt)
} else {
account.AuthSource = "local"
err = tx.QueryRow(ctx, `
UPDATE gateway.portal_users
SET account = $2, name = $3, role = $4, permissions = $5,
active = $6, password_hash = COALESCE($7, password_hash), department_id = $8,
updated_at = clock_timestamp()
WHERE id = $1
RETURNING auth_source, totp_enabled, locked_until, created_at, updated_at`, account.ID, account.Login,
account.DisplayName, account.Role, account.Permissions, account.Active, passwordHash, account.DepartmentID,
).Scan(&account.AuthSource, &account.TOTPEnabled, &account.LockedUntil, &account.CreatedAt, &account.UpdatedAt)
}
if err != nil {
return Account{}, mapManagementError(err)
}
payload, _ := json.Marshal(map[string]any{"identity_id": account.ID, "kind": account.Kind, "actor_id": actorID})
if _, err := tx.Exec(ctx, `
INSERT INTO gateway.outbox_events
(event_id, event_type, event_version, aggregate_type, aggregate_id, payload)
VALUES ($1, 'identity.updated', 1, 'identity', $2, $3)`, eventID, account.ID, payload); err != nil {
return Account{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
if err := tx.Commit(ctx); err != nil {
return Account{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
}
oplog.Record(ctx, r.pool, nil, actorID, account.Login, "identity.update", "identity", account.ID, map[string]any{
"kind": account.Kind, "role": account.Role, "active": account.Active,
"department_id": account.DepartmentID, "password_changed": passwordHash != nil,
})
return account, nil
}
func mapManagementError(err error) error {
if errors.Is(err, pgx.ErrNoRows) {
return ErrNotFound
}
var pgError *pgconn.PgError
if errors.As(err, &pgError) && pgError.Code == "23505" {
return ErrIdentityConflict
}
if err != nil {
return fmt.Errorf("%w: %v", ErrUnavailable, err)
}
return nil
}