feat(server/ops-api): add operations console backend
Stand up an internal ops-api service with operator account, auth, and audit log subsystems backed by a dedicated migrations_ops migration set. Wire Makefile targets and the migrate Dockerfile stage so the new schema ships alongside the existing tenant + monitoring databases. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,194 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/geo-platform/tenant-api/internal/ops/domain"
|
||||
)
|
||||
|
||||
var ErrAccountNotFound = errors.New("operator account not found")
|
||||
|
||||
type AccountRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewAccountRepository(pool *pgxpool.Pool) *AccountRepository {
|
||||
return &AccountRepository{pool: pool}
|
||||
}
|
||||
|
||||
const accountColumns = `id, username, email, password_hash, display_name, role, status,
|
||||
last_login_at, created_by, created_at, updated_at`
|
||||
|
||||
func scanAccount(row pgx.Row) (*domain.OperatorAccount, error) {
|
||||
var a domain.OperatorAccount
|
||||
err := row.Scan(
|
||||
&a.ID, &a.Username, &a.Email, &a.PasswordHash, &a.DisplayName,
|
||||
&a.Role, &a.Status, &a.LastLoginAt, &a.CreatedBy, &a.CreatedAt, &a.UpdatedAt,
|
||||
)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrAccountNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return &a, nil
|
||||
}
|
||||
|
||||
func (r *AccountRepository) GetByUsername(ctx context.Context, username string) (*domain.OperatorAccount, error) {
|
||||
row := r.pool.QueryRow(ctx, `
|
||||
SELECT `+accountColumns+`
|
||||
FROM ops.operator_accounts
|
||||
WHERE username = $1 AND deleted_at IS NULL`, username)
|
||||
return scanAccount(row)
|
||||
}
|
||||
|
||||
func (r *AccountRepository) GetByID(ctx context.Context, id int64) (*domain.OperatorAccount, error) {
|
||||
row := r.pool.QueryRow(ctx, `
|
||||
SELECT `+accountColumns+`
|
||||
FROM ops.operator_accounts
|
||||
WHERE id = $1 AND deleted_at IS NULL`, id)
|
||||
return scanAccount(row)
|
||||
}
|
||||
|
||||
func (r *AccountRepository) Count(ctx context.Context) (int64, error) {
|
||||
var n int64
|
||||
err := r.pool.QueryRow(ctx, `SELECT COUNT(*) FROM ops.operator_accounts WHERE deleted_at IS NULL`).Scan(&n)
|
||||
return n, err
|
||||
}
|
||||
|
||||
type ListFilter struct {
|
||||
Keyword string
|
||||
Status string
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
func (r *AccountRepository) List(ctx context.Context, f ListFilter) ([]domain.OperatorAccount, int64, error) {
|
||||
args := []any{}
|
||||
where := "deleted_at IS NULL"
|
||||
|
||||
if f.Keyword != "" {
|
||||
args = append(args, "%"+f.Keyword+"%")
|
||||
where += fmt.Sprintf(" AND (username ILIKE $%d OR display_name ILIKE $%d OR email ILIKE $%d)",
|
||||
len(args), len(args), len(args))
|
||||
}
|
||||
if f.Status != "" {
|
||||
args = append(args, f.Status)
|
||||
where += fmt.Sprintf(" AND status = $%d", len(args))
|
||||
}
|
||||
|
||||
var total int64
|
||||
if err := r.pool.QueryRow(ctx,
|
||||
"SELECT COUNT(*) FROM ops.operator_accounts WHERE "+where, args...).Scan(&total); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
limit := f.Limit
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 20
|
||||
}
|
||||
offset := f.Offset
|
||||
if offset < 0 {
|
||||
offset = 0
|
||||
}
|
||||
args = append(args, limit, offset)
|
||||
limitArg := len(args) - 1
|
||||
offsetArg := len(args)
|
||||
|
||||
rows, err := r.pool.Query(ctx, fmt.Sprintf(`
|
||||
SELECT %s
|
||||
FROM ops.operator_accounts
|
||||
WHERE %s
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $%d OFFSET $%d`, accountColumns, where, limitArg, offsetArg), args...)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]domain.OperatorAccount, 0, limit)
|
||||
for rows.Next() {
|
||||
var a domain.OperatorAccount
|
||||
if err := rows.Scan(
|
||||
&a.ID, &a.Username, &a.Email, &a.PasswordHash, &a.DisplayName,
|
||||
&a.Role, &a.Status, &a.LastLoginAt, &a.CreatedBy, &a.CreatedAt, &a.UpdatedAt,
|
||||
); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, total, rows.Err()
|
||||
}
|
||||
|
||||
type CreateAccountInput struct {
|
||||
Username string
|
||||
Email *string
|
||||
PasswordHash string
|
||||
DisplayName string
|
||||
Role string
|
||||
CreatedBy *int64
|
||||
}
|
||||
|
||||
func (r *AccountRepository) Create(ctx context.Context, in CreateAccountInput) (*domain.OperatorAccount, error) {
|
||||
row := r.pool.QueryRow(ctx, `
|
||||
INSERT INTO ops.operator_accounts (username, email, password_hash, display_name, role, status, created_by)
|
||||
VALUES ($1, $2, $3, $4, $5, 'active', $6)
|
||||
RETURNING `+accountColumns,
|
||||
in.Username, in.Email, in.PasswordHash, in.DisplayName, in.Role, in.CreatedBy)
|
||||
return scanAccount(row)
|
||||
}
|
||||
|
||||
func (r *AccountRepository) UpdateStatus(ctx context.Context, id int64, status string) error {
|
||||
cmd, err := r.pool.Exec(ctx, `
|
||||
UPDATE ops.operator_accounts
|
||||
SET status = $1, updated_at = NOW()
|
||||
WHERE id = $2 AND deleted_at IS NULL`, status, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if cmd.RowsAffected() == 0 {
|
||||
return ErrAccountNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *AccountRepository) UpdatePassword(ctx context.Context, id int64, passwordHash string) error {
|
||||
cmd, err := r.pool.Exec(ctx, `
|
||||
UPDATE ops.operator_accounts
|
||||
SET password_hash = $1, updated_at = NOW()
|
||||
WHERE id = $2 AND deleted_at IS NULL`, passwordHash, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if cmd.RowsAffected() == 0 {
|
||||
return ErrAccountNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *AccountRepository) UpdateProfile(ctx context.Context, id int64, displayName string, email *string) error {
|
||||
cmd, err := r.pool.Exec(ctx, `
|
||||
UPDATE ops.operator_accounts
|
||||
SET display_name = $1, email = $2, updated_at = NOW()
|
||||
WHERE id = $3 AND deleted_at IS NULL`, displayName, email, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if cmd.RowsAffected() == 0 {
|
||||
return ErrAccountNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *AccountRepository) MarkLogin(ctx context.Context, id int64) error {
|
||||
_, err := r.pool.Exec(ctx, `
|
||||
UPDATE ops.operator_accounts
|
||||
SET last_login_at = NOW(), updated_at = NOW()
|
||||
WHERE id = $1`, id)
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/geo-platform/tenant-api/internal/ops/domain"
|
||||
)
|
||||
|
||||
type AuditRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewAuditRepository(pool *pgxpool.Pool) *AuditRepository {
|
||||
return &AuditRepository{pool: pool}
|
||||
}
|
||||
|
||||
func (r *AuditRepository) Append(ctx context.Context, e domain.AuditEvent) error {
|
||||
var metaBytes []byte
|
||||
if e.Metadata != nil {
|
||||
b, err := json.Marshal(e.Metadata)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal audit metadata: %w", err)
|
||||
}
|
||||
metaBytes = b
|
||||
}
|
||||
|
||||
var (
|
||||
operatorName = nullableString(e.OperatorName)
|
||||
targetType = nullableString(e.TargetType)
|
||||
targetID = nullableString(e.TargetID)
|
||||
ip = nullableString(e.IP)
|
||||
ua = nullableString(e.UserAgent)
|
||||
reqID = nullableString(e.RequestID)
|
||||
)
|
||||
|
||||
_, err := r.pool.Exec(ctx, `
|
||||
INSERT INTO ops.audit_logs
|
||||
(operator_id, operator_name, action, target_type, target_id, metadata, ip, user_agent, request_id)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
|
||||
e.OperatorID, operatorName, e.Action, targetType, targetID, metaBytes, ip, ua, reqID)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *AuditRepository) List(ctx context.Context, f domain.AuditLogFilter) ([]domain.AuditLog, int64, error) {
|
||||
args := []any{}
|
||||
where := "1=1"
|
||||
|
||||
if f.OperatorID != nil {
|
||||
args = append(args, *f.OperatorID)
|
||||
where += fmt.Sprintf(" AND operator_id = $%d", len(args))
|
||||
}
|
||||
if f.Action != "" {
|
||||
args = append(args, f.Action)
|
||||
where += fmt.Sprintf(" AND action = $%d", len(args))
|
||||
}
|
||||
if f.StartAt != nil {
|
||||
args = append(args, *f.StartAt)
|
||||
where += fmt.Sprintf(" AND created_at >= $%d", len(args))
|
||||
}
|
||||
if f.EndAt != nil {
|
||||
args = append(args, *f.EndAt)
|
||||
where += fmt.Sprintf(" AND created_at < $%d", len(args))
|
||||
}
|
||||
|
||||
var total int64
|
||||
if err := r.pool.QueryRow(ctx,
|
||||
"SELECT COUNT(*) FROM ops.audit_logs WHERE "+where, args...).Scan(&total); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
limit := f.Limit
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 50
|
||||
}
|
||||
offset := f.Offset
|
||||
if offset < 0 {
|
||||
offset = 0
|
||||
}
|
||||
args = append(args, limit, offset)
|
||||
limitArg := len(args) - 1
|
||||
offsetArg := len(args)
|
||||
|
||||
rows, err := r.pool.Query(ctx, fmt.Sprintf(`
|
||||
SELECT id, operator_id, operator_name, action, target_type, target_id,
|
||||
metadata, ip, user_agent, request_id, created_at
|
||||
FROM ops.audit_logs
|
||||
WHERE %s
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT $%d OFFSET $%d`, where, limitArg, offsetArg), args...)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]domain.AuditLog, 0, limit)
|
||||
for rows.Next() {
|
||||
var (
|
||||
log domain.AuditLog
|
||||
metaRaw []byte
|
||||
)
|
||||
if err := rows.Scan(
|
||||
&log.ID, &log.OperatorID, &log.OperatorName, &log.Action,
|
||||
&log.TargetType, &log.TargetID, &metaRaw, &log.IP, &log.UserAgent,
|
||||
&log.RequestID, &log.CreatedAt,
|
||||
); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if len(metaRaw) > 0 {
|
||||
if err := json.Unmarshal(metaRaw, &log.Metadata); err != nil {
|
||||
return nil, 0, fmt.Errorf("decode audit metadata: %w", err)
|
||||
}
|
||||
}
|
||||
out = append(out, log)
|
||||
}
|
||||
return out, total, rows.Err()
|
||||
}
|
||||
|
||||
func nullableString(s string) *string {
|
||||
if s == "" {
|
||||
return nil
|
||||
}
|
||||
return &s
|
||||
}
|
||||
Reference in New Issue
Block a user