ea78ef5674
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 干净。
633 lines
23 KiB
Go
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
|
|
}
|