goclaw/internal/store/pg/activity_store.go

package pg

import (
	"context"
	"database/sql"
	"fmt"
	"strings"

	"github.com/google/uuid"

	"github.com/nextlevelbuilder/goclaw/internal/store"
)

// PGActivityStore implements store.ActivityStore backed by Postgres.
type PGActivityStore struct {
	db *sql.DB
}

// NewPGActivityStore creates a new PGActivityStore.
func NewPGActivityStore(db *sql.DB) *PGActivityStore {
	return &PGActivityStore{db: db}
}

func (s *PGActivityStore) Log(ctx context.Context, entry *store.ActivityLog) error {
	tenantID := store.TenantIDFromContext(ctx)
	if tenantID == uuid.Nil {
		tenantID = store.MasterTenantID
	}
	_, err := s.db.ExecContext(ctx,
		`INSERT INTO activity_logs (actor_type, actor_id, action, entity_type, entity_id, details, ip_address, tenant_id)
		 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
		entry.ActorType, entry.ActorID, entry.Action,
		entry.EntityType, entry.EntityID, entry.Details, entry.IPAddress, tenantID,
	)
	return err
}

func (s *PGActivityStore) List(ctx context.Context, opts store.ActivityListOpts) ([]store.ActivityLog, error) {
	where, args := buildActivityWhere(ctx, opts)
	limit := opts.Limit
	if limit <= 0 {
		limit = 50
	}
	args = append(args, limit, opts.Offset)

	query := fmt.Sprintf(
		`SELECT id, actor_type, actor_id, action, COALESCE(entity_type,''), COALESCE(entity_id,''), COALESCE(details, 'null'::jsonb), COALESCE(ip_address,''), created_at
		 FROM activity_logs %s ORDER BY created_at DESC LIMIT $%d OFFSET $%d`,
		where, len(args)-1, len(args),
	)

	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var result []store.ActivityLog
	for rows.Next() {
		var a store.ActivityLog
		if err := rows.Scan(&a.ID, &a.ActorType, &a.ActorID, &a.Action, &a.EntityType, &a.EntityID, &a.Details, &a.IPAddress, &a.CreatedAt); err != nil {
			return nil, err
		}
		result = append(result, a)
	}
	return result, rows.Err()
}

func (s *PGActivityStore) Count(ctx context.Context, opts store.ActivityListOpts) (int, error) {
	where, args := buildActivityWhere(ctx, opts)
	query := fmt.Sprintf("SELECT COUNT(*) FROM activity_logs %s", where)

	var count int
	err := s.db.QueryRowContext(ctx, query, args...).Scan(&count)
	return count, err
}

func (s *PGActivityStore) Aggregate(ctx context.Context, opts store.ActivityAggregateOpts) ([]store.ActivityAggregateBucket, int, error) {
	column, ok := activityAggregateColumn(opts.GroupBy)
	if !ok {
		return nil, 0, fmt.Errorf("invalid group_by")
	}
	limit := opts.Limit
	if limit <= 0 {
		limit = 50
	}
	where, args := buildActivityWhere(ctx, opts.ActivityListOpts)
	total, err := s.Count(ctx, opts.ActivityListOpts)
	if err != nil {
		return nil, 0, err
	}
	args = append(args, limit)
	query := fmt.Sprintf(
		`SELECT COALESCE(%s, ''), COUNT(*), MAX(created_at)
		 FROM activity_logs %s
		 GROUP BY %s
		 ORDER BY COUNT(*) DESC, MAX(created_at) DESC
		 LIMIT $%d`,
		column, where, column, len(args),
	)
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()

	var buckets []store.ActivityAggregateBucket
	for rows.Next() {
		var b store.ActivityAggregateBucket
		if err := rows.Scan(&b.Key, &b.Count, &b.LastSeen); err != nil {
			return nil, 0, err
		}
		buckets = append(buckets, b)
	}
	return buckets, total, rows.Err()
}

func buildActivityWhere(ctx context.Context, opts store.ActivityListOpts) (string, []any) {
	var conditions []string
	var args []any
	idx := 1

	if !store.IsCrossTenant(ctx) {
		tenantID := store.TenantIDFromContext(ctx)
		if tenantID != uuid.Nil {
			conditions = append(conditions, fmt.Sprintf("tenant_id = $%d", idx))
			args = append(args, tenantID)
			idx++
		}
	}

	if opts.ActorType != "" {
		conditions = append(conditions, fmt.Sprintf("actor_type = $%d", idx))
		args = append(args, opts.ActorType)
		idx++
	}
	if opts.ActorID != "" {
		conditions = append(conditions, fmt.Sprintf("actor_id = $%d", idx))
		args = append(args, opts.ActorID)
		idx++
	}
	if opts.Action != "" {
		conditions = append(conditions, fmt.Sprintf("action = $%d", idx))
		args = append(args, opts.Action)
		idx++
	}
	if opts.EntityType != "" {
		conditions = append(conditions, fmt.Sprintf("entity_type = $%d", idx))
		args = append(args, opts.EntityType)
		idx++
	}
	if opts.EntityID != "" {
		conditions = append(conditions, fmt.Sprintf("entity_id = $%d", idx))
		args = append(args, opts.EntityID)
		idx++
	}
	if opts.From != nil {
		conditions = append(conditions, fmt.Sprintf("created_at >= $%d", idx))
		args = append(args, *opts.From)
		idx++
	}
	if opts.To != nil {
		conditions = append(conditions, fmt.Sprintf("created_at < $%d", idx))
		args = append(args, *opts.To)
		idx++
	}

	if len(conditions) == 0 {
		return "", nil
	}
	return "WHERE " + strings.Join(conditions, " AND "), args
}

func activityAggregateColumn(groupBy string) (string, bool) {
	switch groupBy {
	case "action":
		return "action", true
	case "actor_type":
		return "actor_type", true
	case "entity_type":
		return "entity_type", true
	case "actor_id":
		return "actor_id", true
	default:
		return "", false
	}
}