lyon
271 lines
12 KiB
Go
271 lines
12 KiB
Go
package userbilling
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/csv"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func (s *Server) handleAdminUsage(w http.ResponseWriter, r *http.Request) {
|
|
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
|
|
return
|
|
}
|
|
query := parseAdminAuditQuery(r)
|
|
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
|
|
defer cancel()
|
|
rows, total, err := s.adminUsageRows(ctx, query)
|
|
if err != nil {
|
|
s.logDatabaseError("admin_usage", err)
|
|
writeAPIError(w, http.StatusInternalServerError, "admin_usage_failed", "could not load usage records")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-usage-v1", "rows": rows, "count": len(rows), "total": total, "pagination": pagination(query.Page, query.PageSize, total), "filters": auditFilterSummary(query), "valuesRedacted": true})
|
|
}
|
|
|
|
func (s *Server) handleAdminUsageExport(w http.ResponseWriter, r *http.Request) {
|
|
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
|
|
return
|
|
}
|
|
query := parseAdminAuditQuery(r)
|
|
query.Page = 1
|
|
query.PageSize = queryLimit(r, "limit", 1000, 5000)
|
|
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
|
defer cancel()
|
|
rows, _, err := s.adminUsageRows(ctx, query)
|
|
if err != nil {
|
|
s.logDatabaseError("admin_usage_export", err)
|
|
writeAPIError(w, http.StatusInternalServerError, "admin_usage_export_failed", "could not export usage records")
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/csv; charset=utf-8")
|
|
w.Header().Set("Content-Disposition", `attachment; filename="hwlab-admin-usage.csv"`)
|
|
writer := csv.NewWriter(w)
|
|
_ = writer.Write([]string{"id", "created_at", "user_id", "email", "username", "api_key_id", "service_id", "resource_type", "unit", "quantity", "credits", "status", "idempotency_key", "metadata_json"})
|
|
for _, row := range rows {
|
|
_ = writer.Write([]string{row.ID, row.CreatedAt.Format(time.RFC3339), row.UserID, row.Email, row.Username, row.APIKeyID, row.ServiceID, row.ResourceType, row.Unit, strconv.FormatInt(row.Quantity, 10), strconv.FormatInt(row.Credits, 10), row.Status, row.IdempotencyKey, compactJSON(row.Metadata)})
|
|
}
|
|
writer.Flush()
|
|
}
|
|
|
|
func (s *Server) handleAdminLedger(w http.ResponseWriter, r *http.Request) {
|
|
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
|
|
return
|
|
}
|
|
query := parseAdminAuditQuery(r)
|
|
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
|
|
defer cancel()
|
|
rows, total, err := s.adminLedgerRows(ctx, query)
|
|
if err != nil {
|
|
s.logDatabaseError("admin_ledger", err)
|
|
writeAPIError(w, http.StatusInternalServerError, "admin_ledger_failed", "could not load ledger rows")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"contractVersion": "user-billing-admin-ledger-v1", "rows": rows, "count": len(rows), "total": total, "pagination": pagination(query.Page, query.PageSize, total), "filters": auditFilterSummary(query), "valuesRedacted": true})
|
|
}
|
|
|
|
func (s *Server) handleAdminLedgerExport(w http.ResponseWriter, r *http.Request) {
|
|
if !s.internalAllowed(w, r) || !s.databaseAvailable(w) {
|
|
return
|
|
}
|
|
query := parseAdminAuditQuery(r)
|
|
query.Page = 1
|
|
query.PageSize = queryLimit(r, "limit", 1000, 5000)
|
|
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
|
defer cancel()
|
|
rows, _, err := s.adminLedgerRows(ctx, query)
|
|
if err != nil {
|
|
s.logDatabaseError("admin_ledger_export", err)
|
|
writeAPIError(w, http.StatusInternalServerError, "admin_ledger_export_failed", "could not export ledger rows")
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/csv; charset=utf-8")
|
|
w.Header().Set("Content-Disposition", `attachment; filename="hwlab-admin-ledger.csv"`)
|
|
writer := csv.NewWriter(w)
|
|
_ = writer.Write([]string{"id", "created_at", "user_id", "email", "username", "kind", "reason", "source", "status", "delta_credits", "balance_before", "balance_after", "operator_user_id", "operator_username", "idempotency_key", "metadata_json"})
|
|
for _, row := range rows {
|
|
before := ""
|
|
if row.BalanceBefore != nil {
|
|
before = strconv.FormatInt(*row.BalanceBefore, 10)
|
|
}
|
|
_ = writer.Write([]string{row.ID, row.CreatedAt.Format(time.RFC3339), row.UserID, row.Email, row.Username, row.Kind, row.Reason, row.Source, row.Status, strconv.FormatInt(row.DeltaCredits, 10), before, strconv.FormatInt(row.BalanceAfter, 10), row.OperatorUserID, row.OperatorUsername, row.IdempotencyKey, compactJSON(row.Metadata)})
|
|
}
|
|
writer.Flush()
|
|
}
|
|
|
|
func (s *Server) adminUsageRows(ctx context.Context, query adminAuditQuery) ([]adminUsageRow, int64, error) {
|
|
where, args := adminUsageWhere(query)
|
|
total, err := countRows(ctx, s.db, `SELECT COUNT(*) FROM hwlab_usage_records ur JOIN hwlab_users u ON u.id = ur.user_id `+where, args)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
rowArgs := append([]any{}, args...)
|
|
rowArgs = append(rowArgs, query.PageSize, (query.Page-1)*query.PageSize)
|
|
limitParam := "$" + strconv.Itoa(len(rowArgs)-1)
|
|
offsetParam := "$" + strconv.Itoa(len(rowArgs))
|
|
rows, err := s.db.QueryContext(ctx, `SELECT ur.id, COALESCE(ur.reservation_id, ''), ur.user_id, u.email, u.username, COALESCE(ur.api_key_id, ''), ur.service_id, ur.resource_type, ur.unit, ur.quantity, ur.credits, ur.status, COALESCE(ur.idempotency_key, ''), ur.metadata::text, ur.created_at FROM hwlab_usage_records ur JOIN hwlab_users u ON u.id = ur.user_id `+where+` ORDER BY ur.created_at DESC, ur.id DESC LIMIT `+limitParam+` OFFSET `+offsetParam, rowArgs...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
items := []adminUsageRow{}
|
|
for rows.Next() {
|
|
var item adminUsageRow
|
|
var metadataRaw string
|
|
if err := rows.Scan(&item.ID, &item.ReservationID, &item.UserID, &item.Email, &item.Username, &item.APIKeyID, &item.ServiceID, &item.ResourceType, &item.Unit, &item.Quantity, &item.Credits, &item.Status, &item.IdempotencyKey, &metadataRaw, &item.CreatedAt); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
item.Metadata = redactedMetadata(parseJSONMap(metadataRaw))
|
|
items = append(items, item)
|
|
}
|
|
return items, total, rows.Err()
|
|
}
|
|
|
|
func (s *Server) adminLedgerRows(ctx context.Context, query adminAuditQuery) ([]adminLedgerRow, int64, error) {
|
|
where, args := adminLedgerWhere(query)
|
|
total, err := countRows(ctx, s.db, `SELECT COUNT(*) FROM hwlab_credit_ledger l JOIN hwlab_users u ON u.id = l.user_id `+where, args)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
rowArgs := append([]any{}, args...)
|
|
rowArgs = append(rowArgs, query.PageSize, (query.Page-1)*query.PageSize)
|
|
limitParam := "$" + strconv.Itoa(len(rowArgs)-1)
|
|
offsetParam := "$" + strconv.Itoa(len(rowArgs))
|
|
rows, err := s.db.QueryContext(ctx, `SELECT l.id, l.user_id, u.email, u.username, l.kind, l.reason, l.delta_credits, l.balance_before, l.balance_after, l.source, l.status, COALESCE(l.idempotency_key, ''), COALESCE(l.operator_user_id, ''), COALESCE(l.operator_username, ''), l.metadata::text, l.created_at FROM hwlab_credit_ledger l JOIN hwlab_users u ON u.id = l.user_id `+where+` ORDER BY l.created_at DESC, l.id DESC LIMIT `+limitParam+` OFFSET `+offsetParam, rowArgs...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
items := []adminLedgerRow{}
|
|
for rows.Next() {
|
|
var item adminLedgerRow
|
|
var before sql.NullInt64
|
|
var metadataRaw string
|
|
if err := rows.Scan(&item.ID, &item.UserID, &item.Email, &item.Username, &item.Kind, &item.Reason, &item.DeltaCredits, &before, &item.BalanceAfter, &item.Source, &item.Status, &item.IdempotencyKey, &item.OperatorUserID, &item.OperatorUsername, &metadataRaw, &item.CreatedAt); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
if before.Valid {
|
|
value := before.Int64
|
|
item.BalanceBefore = &value
|
|
}
|
|
item.Metadata = redactedMetadata(parseJSONMap(metadataRaw))
|
|
items = append(items, item)
|
|
}
|
|
return items, total, rows.Err()
|
|
}
|
|
|
|
func adminUsageWhere(query adminAuditQuery) (string, []any) {
|
|
clauses, args := []string{}, []any{}
|
|
addAuditClause(&clauses, &args, "ur.user_id = %s", query.UserID)
|
|
addAuditClause(&clauses, &args, "ur.service_id = %s", query.ServiceID)
|
|
addAuditClause(&clauses, &args, "ur.resource_type = %s", query.ResourceType)
|
|
addAuditClause(&clauses, &args, "ur.api_key_id = %s", query.APIKeyID)
|
|
addAuditClause(&clauses, &args, "ur.status = %s", query.Status)
|
|
addAuditClause(&clauses, &args, "ur.idempotency_key = %s", query.IdempotencyKey)
|
|
addAuditTimeClause(&clauses, &args, "ur.created_at >= %s::timestamptz", query.From)
|
|
addAuditTimeClause(&clauses, &args, "ur.created_at <= %s::timestamptz", query.To)
|
|
if search := strings.TrimSpace(strings.ToLower(query.Search)); search != "" {
|
|
args = append(args, "%"+search+"%")
|
|
param := "$" + strconv.Itoa(len(args))
|
|
clauses = append(clauses, "(lower(ur.id) LIKE "+param+" OR lower(ur.service_id) LIKE "+param+" OR lower(ur.resource_type) LIKE "+param+" OR lower(u.email) LIKE "+param+" OR lower(u.username) LIKE "+param+")")
|
|
}
|
|
return whereSQL(clauses), args
|
|
}
|
|
|
|
func adminLedgerWhere(query adminAuditQuery) (string, []any) {
|
|
clauses, args := []string{}, []any{}
|
|
addAuditClause(&clauses, &args, "l.user_id = %s", query.UserID)
|
|
addAuditClause(&clauses, &args, "l.kind = %s", query.Kind)
|
|
addAuditClause(&clauses, &args, "l.source = %s", query.Source)
|
|
addAuditClause(&clauses, &args, "l.status = %s", query.Status)
|
|
addAuditClause(&clauses, &args, "l.idempotency_key = %s", query.IdempotencyKey)
|
|
addAuditTimeClause(&clauses, &args, "l.created_at >= %s::timestamptz", query.From)
|
|
addAuditTimeClause(&clauses, &args, "l.created_at <= %s::timestamptz", query.To)
|
|
if search := strings.TrimSpace(strings.ToLower(query.Search)); search != "" {
|
|
args = append(args, "%"+search+"%")
|
|
param := "$" + strconv.Itoa(len(args))
|
|
clauses = append(clauses, "(lower(l.id) LIKE "+param+" OR lower(l.kind) LIKE "+param+" OR lower(l.reason) LIKE "+param+" OR lower(u.email) LIKE "+param+" OR lower(u.username) LIKE "+param+")")
|
|
}
|
|
return whereSQL(clauses), args
|
|
}
|
|
|
|
func parseAdminAuditQuery(r *http.Request) adminAuditQuery {
|
|
return adminAuditQuery{Page: queryLimit(r, "page", 1, 100000), PageSize: queryLimit(r, "pageSize", 50, 500), UserID: textQuery(r, "userId"), Search: textQuery(r, "search"), ServiceID: textQuery(r, "serviceId"), ResourceType: textQuery(r, "resourceType"), APIKeyID: textQuery(r, "apiKeyId"), Status: textQuery(r, "status"), Kind: textQuery(r, "kind"), Source: textQuery(r, "source"), IdempotencyKey: textQuery(r, "idempotencyKey"), From: textQuery(r, "from"), To: textQuery(r, "to")}
|
|
}
|
|
|
|
func addAuditClause(clauses *[]string, args *[]any, template, value string) {
|
|
if value = strings.TrimSpace(value); value != "" {
|
|
*args = append(*args, value)
|
|
*clauses = append(*clauses, fmt.Sprintf(template, "$"+strconv.Itoa(len(*args))))
|
|
}
|
|
}
|
|
|
|
func addAuditTimeClause(clauses *[]string, args *[]any, template, value string) {
|
|
addAuditClause(clauses, args, template, value)
|
|
}
|
|
|
|
func whereSQL(clauses []string) string {
|
|
if len(clauses) == 0 {
|
|
return ""
|
|
}
|
|
return "WHERE " + strings.Join(clauses, " AND ")
|
|
}
|
|
|
|
func countRows(ctx context.Context, db *sql.DB, query string, args []any) (int64, error) {
|
|
var total int64
|
|
err := db.QueryRowContext(ctx, query, args...).Scan(&total)
|
|
return total, err
|
|
}
|
|
|
|
func pagination(page, pageSize int, total int64) map[string]any {
|
|
totalPages := int64(1)
|
|
if pageSize > 0 && total > 0 {
|
|
totalPages = (total + int64(pageSize) - 1) / int64(pageSize)
|
|
}
|
|
return map[string]any{"page": page, "pageSize": pageSize, "total": total, "totalPages": totalPages}
|
|
}
|
|
|
|
func auditFilterSummary(query adminAuditQuery) map[string]any {
|
|
return map[string]any{"userId": query.UserID, "search": query.Search, "serviceId": query.ServiceID, "resourceType": query.ResourceType, "apiKeyId": query.APIKeyID, "status": query.Status, "kind": query.Kind, "source": query.Source, "idempotencyKey": query.IdempotencyKey, "from": query.From, "to": query.To}
|
|
}
|
|
|
|
func textQuery(r *http.Request, name string) string {
|
|
return strings.TrimSpace(r.URL.Query().Get(name))
|
|
}
|
|
|
|
func compactJSON(value map[string]any) string {
|
|
if len(value) == 0 {
|
|
return "{}"
|
|
}
|
|
body, err := json.Marshal(value)
|
|
if err != nil {
|
|
return "{}"
|
|
}
|
|
return string(body)
|
|
}
|
|
|
|
func redactedMetadata(input map[string]any) map[string]any {
|
|
if len(input) == 0 {
|
|
return map[string]any{}
|
|
}
|
|
output := map[string]any{}
|
|
for key, value := range input {
|
|
lower := strings.ToLower(key)
|
|
if strings.Contains(lower, "token") || strings.Contains(lower, "secret") || strings.Contains(lower, "password") || strings.Contains(lower, "database_url") || strings.Contains(lower, "dsn") || strings.Contains(lower, "key_hash") || strings.Contains(lower, "apikey") || strings.Contains(lower, "api_key") {
|
|
output[key] = "redacted"
|
|
continue
|
|
}
|
|
if nested, ok := value.(map[string]any); ok {
|
|
output[key] = redactedMetadata(nested)
|
|
continue
|
|
}
|
|
output[key] = value
|
|
}
|
|
return output
|
|
}
|