MengMengCode_VoCat_20260809-175859/internal/store/events.go

package store

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"
)

func (s *Store) AppendAuditEvent(ctx context.Context, value AuditEvent) (AuditEvent, error) {
	value.Action = strings.TrimSpace(value.Action)
	if value.Action == "" {
		return AuditEvent{}, errors.New("audit action is required")
	}
	details, err := normalizeJSONObject(value.Details)
	if err != nil {
		return AuditEvent{}, fmt.Errorf("normalize audit details: %w", err)
	}
	if value.CreatedAt.IsZero() {
		value.CreatedAt = time.Now().UTC()
	}
	result, err := s.db.ExecContext(ctx, `
		INSERT INTO audit_events (
			actor, action, entity_type, entity_id, outcome, remote_addr,
			details_json, created_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	`,
		value.Actor, value.Action, value.EntityType, value.EntityID,
		value.Outcome, value.RemoteAddr, string(details), value.CreatedAt.Unix(),
	)
	if err != nil {
		return AuditEvent{}, fmt.Errorf("append audit event: %w", err)
	}
	value.ID, err = result.LastInsertId()
	if err != nil {
		return AuditEvent{}, fmt.Errorf("read audit event id: %w", err)
	}
	value.Details = details
	return value, nil
}

func (s *Store) ListAuditEvents(ctx context.Context, filter AuditFilter) ([]AuditEvent, error) {
	clauses := make([]string, 0, 7)
	args := make([]any, 0, 8)
	if filter.Actor != "" {
		clauses = append(clauses, `actor = ?`)
		args = append(args, filter.Actor)
	}
	if filter.Action != "" {
		clauses = append(clauses, `action = ?`)
		args = append(args, filter.Action)
	}
	if filter.EntityType != "" {
		clauses = append(clauses, `entity_type = ?`)
		args = append(args, filter.EntityType)
	}
	if filter.EntityID != "" {
		clauses = append(clauses, `entity_id = ?`)
		args = append(args, filter.EntityID)
	}
	if !filter.Since.IsZero() {
		clauses = append(clauses, `created_at >= ?`)
		args = append(args, filter.Since.UTC().Unix())
	}
	if !filter.Until.IsZero() {
		clauses = append(clauses, `created_at < ?`)
		args = append(args, filter.Until.UTC().Unix())
	}
	if filter.BeforeID > 0 {
		clauses = append(clauses, `id < ?`)
		args = append(args, filter.BeforeID)
	}
	query := auditEventSelect
	if len(clauses) > 0 {
		query += ` WHERE ` + strings.Join(clauses, ` AND `)
	}
	query += ` ORDER BY created_at DESC, id DESC LIMIT ?`
	args = append(args, normalizedLimit(filter.Limit))
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list audit events: %w", err)
	}
	defer rows.Close()
	values := make([]AuditEvent, 0)
	for rows.Next() {
		value, err := auditEvent(rows)
		if err != nil {
			return nil, fmt.Errorf("scan audit event: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate audit events: %w", err)
	}
	return values, nil
}

const auditEventSelect = `
	SELECT id, actor, action, entity_type, entity_id, outcome,
		remote_addr, details_json, created_at
	FROM audit_events`

func auditEvent(row rowScanner) (AuditEvent, error) {
	var value AuditEvent
	var details string
	var createdAt int64
	err := row.Scan(
		&value.ID, &value.Actor, &value.Action, &value.EntityType,
		&value.EntityID, &value.Outcome, &value.RemoteAddr, &details,
		&createdAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return AuditEvent{}, ErrNotFound
	}
	if err != nil {
		return AuditEvent{}, err
	}
	value.Details = []byte(details)
	value.CreatedAt = time.Unix(createdAt, 0).UTC()
	return value, nil
}

func (s *Store) AppendLogEvent(ctx context.Context, value LogEvent) (LogEvent, error) {
	value.Level = strings.ToLower(strings.TrimSpace(value.Level))
	if value.Level == "" {
		return LogEvent{}, errors.New("log level is required")
	}
	if strings.TrimSpace(value.Message) == "" {
		return LogEvent{}, errors.New("log message is required")
	}
	fields, err := normalizeJSONValue(value.Fields)
	if err != nil {
		return LogEvent{}, fmt.Errorf("normalize log fields: %w", err)
	}
	if value.Time.IsZero() {
		value.Time = time.Now().UTC()
	}
	result, err := s.db.ExecContext(ctx, `
		INSERT INTO log_events (event_time, level, message, caller, fields_json)
		VALUES (?, ?, ?, ?, ?)
	`, value.Time.Unix(), value.Level, value.Message, value.Caller, string(fields))
	if err != nil {
		return LogEvent{}, fmt.Errorf("append log event: %w", err)
	}
	value.ID, err = result.LastInsertId()
	if err != nil {
		return LogEvent{}, fmt.Errorf("read log event id: %w", err)
	}
	value.Fields = fields
	return value, nil
}

func (s *Store) ListLogEvents(ctx context.Context, filter LogFilter) ([]LogEvent, error) {
	clauses := make([]string, 0, 4)
	args := make([]any, 0, 5)
	if filter.Level != "" {
		clauses = append(clauses, `level = ?`)
		args = append(args, strings.ToLower(filter.Level))
	}
	if !filter.Since.IsZero() {
		clauses = append(clauses, `event_time >= ?`)
		args = append(args, filter.Since.UTC().Unix())
	}
	if !filter.Until.IsZero() {
		clauses = append(clauses, `event_time < ?`)
		args = append(args, filter.Until.UTC().Unix())
	}
	if filter.BeforeID > 0 {
		clauses = append(clauses, `id < ?`)
		args = append(args, filter.BeforeID)
	}
	query := logEventSelect
	if len(clauses) > 0 {
		query += ` WHERE ` + strings.Join(clauses, ` AND `)
	}
	query += ` ORDER BY event_time DESC, id DESC LIMIT ?`
	args = append(args, normalizedLimit(filter.Limit))
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list log events: %w", err)
	}
	defer rows.Close()
	values := make([]LogEvent, 0)
	for rows.Next() {
		value, err := logEvent(rows)
		if err != nil {
			return nil, fmt.Errorf("scan log event: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate log events: %w", err)
	}
	return values, nil
}

const logEventSelect = `
	SELECT id, event_time, level, message, caller, fields_json
	FROM log_events`

func logEvent(row rowScanner) (LogEvent, error) {
	var value LogEvent
	var eventTime int64
	var fields string
	err := row.Scan(
		&value.ID, &eventTime, &value.Level, &value.Message,
		&value.Caller, &fields,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return LogEvent{}, ErrNotFound
	}
	if err != nil {
		return LogEvent{}, err
	}
	value.Time = time.Unix(eventTime, 0).UTC()
	value.Fields = []byte(fields)
	return value, nil
}

func (s *Store) PruneAuditEvents(ctx context.Context, before time.Time) (int64, error) {
	return deleteEventsBefore(ctx, s.db, `audit_events`, `created_at`, before)
}

func (s *Store) PruneLogEvents(ctx context.Context, before time.Time) (int64, error) {
	return deleteEventsBefore(ctx, s.db, `log_events`, `event_time`, before)
}

// CountLogEvents returns how many application log rows are persisted.
func (s *Store) CountLogEvents(ctx context.Context) (int64, error) {
	var count int64
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM log_events`).Scan(&count); err != nil {
		return 0, fmt.Errorf("count log events: %w", err)
	}
	return count, nil
}

// PruneLogEventsToCount keeps only the newest `keep` log rows, deleting the
// rest. keep <= 0 deletes everything.
func (s *Store) PruneLogEventsToCount(ctx context.Context, keep int) (int64, error) {
	if keep < 0 {
		keep = 0
	}
	result, err := s.db.ExecContext(ctx, `
		DELETE FROM log_events WHERE id NOT IN (
			SELECT id FROM log_events ORDER BY id DESC LIMIT ?
		)
	`, keep)
	if err != nil {
		return 0, fmt.Errorf("prune log events to count: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("read pruned log count: %w", err)
	}
	return affected, nil
}

func deleteEventsBefore(
	ctx context.Context,
	executor contextExecer,
	table string,
	column string,
	before time.Time,
) (int64, error) {
	// table and column are internal constants from the callers above.
	result, err := executor.ExecContext(
		ctx,
		`DELETE FROM `+table+` WHERE `+column+` < ?`,
		before.UTC().Unix(),
	)
	if err != nil {
		return 0, fmt.Errorf("prune %s: %w", table, err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("read pruned %s count: %w", table, err)
	}
	return affected, nil
}

// PruneEvents removes audit and application logs atomically.
func (s *Store) PruneEvents(
	ctx context.Context,
	auditBefore time.Time,
	logBefore time.Time,
) (auditCount int64, logCount int64, err error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, 0, fmt.Errorf("begin event pruning: %w", err)
	}
	defer tx.Rollback()
	auditCount, err = deleteEventsBefore(ctx, tx, `audit_events`, `created_at`, auditBefore)
	if err != nil {
		return 0, 0, err
	}
	logCount, err = deleteEventsBefore(ctx, tx, `log_events`, `event_time`, logBefore)
	if err != nil {
		return 0, 0, err
	}
	if err := tx.Commit(); err != nil {
		return 0, 0, fmt.Errorf("commit event pruning: %w", err)
	}
	return auditCount, logCount, nil
}