databasus--databasus/backend/internal/features/telemetry/service.go

package telemetry

import (
	"context"
	"log/slog"
	"math"
	"runtime"
	"sort"
	"time"

	"github.com/google/uuid"

	backups_core_logical "databasus-backend/internal/features/backups/backups/core/logical"
	physical_models "databasus-backend/internal/features/backups/backups/core/physical/models"
	"databasus-backend/internal/features/databases"
	"databasus-backend/internal/features/notifiers"
	"databasus-backend/internal/features/storages"
	verification_agents "databasus-backend/internal/features/verification/agents"
	verification_config "databasus-backend/internal/features/verification/config"
)

const (
	// activeBackupWindow is how far back a successful backup must have happened
	// for a database with disabled healthcheck to count as "active".
	activeBackupWindow = 7 * 24 * time.Hour

	// maxArrayEntries matches the server-side cap from TELEMETRY.md.
	maxArrayEntries = 200
)

type databaseLister interface {
	GetAllDatabases() ([]*databases.Database, error)
}

type storageLister interface {
	GetAllStorages() ([]*storages.Storage, error)
}

type notifierLister interface {
	GetAllNotifiers() ([]*notifiers.Notifier, error)
}

type backupChecker interface {
	HasSuccessfulBackupSince(databaseID uuid.UUID, since time.Time) (bool, error)
	GetLatestCompletedBackup(databaseID uuid.UUID) (*backups_core_logical.LogicalBackup, error)
}

type physicalFullBackupSizer interface {
	GetLatestCompletedFullBackup(databaseID uuid.UUID) (*physical_models.PhysicalFullBackup, error)
}

type userCounter interface {
	GetUsersCount() (int64, error)
}

type verificationAgentLister interface {
	ListAgents() ([]*verification_agents.Agent, error)
}

type verificationConfigLister interface {
	ListEnabled() ([]*verification_config.BackupVerificationConfig, error)
}

type TelemetryService struct {
	instanceLoader            *InstanceFileLoader
	sender                    TelemetrySender
	databaseService           databaseLister
	storageService            storageLister
	notifierService           notifierLister
	backupService             backupChecker
	physicalBackupService     physicalFullBackupSizer
	verificationAgentService  verificationAgentLister
	verificationConfigService verificationConfigLister
	userService               userCounter
	appVersion                string
	logger                    *slog.Logger
}

func NewTelemetryService(
	instanceLoader *InstanceFileLoader,
	sender TelemetrySender,
	databaseService databaseLister,
	storageService storageLister,
	notifierService notifierLister,
	backupService backupChecker,
	physicalBackupService physicalFullBackupSizer,
	verificationAgentService verificationAgentLister,
	verificationConfigService verificationConfigLister,
	userService userCounter,
	appVersion string,
	logger *slog.Logger,
) *TelemetryService {
	return &TelemetryService{
		instanceLoader:            instanceLoader,
		sender:                    sender,
		databaseService:           databaseService,
		storageService:            storageService,
		notifierService:           notifierService,
		backupService:             backupService,
		physicalBackupService:     physicalBackupService,
		verificationAgentService:  verificationAgentService,
		verificationConfigService: verificationConfigService,
		userService:               userService,
		appVersion:                appVersion,
		logger:                    logger,
	}
}

func (s *TelemetryService) BuildAndSend(ctx context.Context) error {
	instance, ok := s.instanceLoader.LoadOrCreate()
	if !ok {
		return nil
	}

	enabledConfigsByDatabaseID, err := s.loadEnabledVerificationConfigs()
	if err != nil {
		return err
	}

	databaseEntries, err := s.collectActiveDatabases(enabledConfigsByDatabaseID)
	if err != nil {
		return err
	}

	storageTypes, err := s.collectStorageTypes()
	if err != nil {
		return err
	}

	notifierTypes, err := s.collectNotifierTypes()
	if err != nil {
		return err
	}

	verificationAgents, err := s.collectVerificationAgents()
	if err != nil {
		return err
	}

	userCount, err := s.userService.GetUsersCount()
	if err != nil {
		return err
	}

	req := &CollectRequest{
		InstanceID:         instance.InstanceID,
		AppVersion:         s.appVersion,
		OS:                 runtime.GOOS,
		Arch:               runtime.GOARCH,
		InstalledAt:        instance.InstalledAt,
		UserCount:          int(userCount),
		Databases:          capDatabases(databaseEntries),
		Storages:           capStrings(storageTypes),
		Notifiers:          capStrings(notifierTypes),
		VerificationAgents: capAgents(verificationAgents),
	}

	return s.sender.Send(ctx, req)
}

func (s *TelemetryService) loadEnabledVerificationConfigs() (
	map[uuid.UUID]*verification_config.BackupVerificationConfig,
	error,
) {
	enabledConfigs, err := s.verificationConfigService.ListEnabled()
	if err != nil {
		return nil, err
	}

	indexed := make(map[uuid.UUID]*verification_config.BackupVerificationConfig, len(enabledConfigs))
	for _, config := range enabledConfigs {
		indexed[config.DatabaseID] = config
	}

	return indexed, nil
}

func (s *TelemetryService) collectActiveDatabases(
	enabledConfigsByDatabaseID map[uuid.UUID]*verification_config.BackupVerificationConfig,
) ([]DatabaseEntry, error) {
	allDatabases, err := s.databaseService.GetAllDatabases()
	if err != nil {
		return nil, err
	}

	since := time.Now().UTC().Add(-activeBackupWindow)
	entries := make([]DatabaseEntry, 0, len(allDatabases))

	for _, db := range allDatabases {
		isActive, err := s.isDatabaseActive(db, since)
		if err != nil {
			return nil, err
		}

		if !isActive {
			continue
		}

		entry, ok := buildDatabaseEntry(db)
		if !ok {
			continue
		}

		if err := s.attachBackupSizes(&entry, db); err != nil {
			return nil, err
		}

		if config, hasConfig := enabledConfigsByDatabaseID[db.ID]; hasConfig {
			entry.Verification = buildVerificationEntry(config)
		}

		entries = append(entries, entry)
	}

	return entries, nil
}

func buildVerificationEntry(
	config *verification_config.BackupVerificationConfig,
) *DatabaseVerificationEntry {
	entry := &DatabaseVerificationEntry{
		IsEnabled:    true,
		ScheduleType: string(config.ScheduleType),
	}

	if config.ScheduleType == verification_config.VerificationScheduleInterval {
		entry.IntervalType = string(config.VerificationInterval.Type)
	}

	return entry
}

func (s *TelemetryService) collectVerificationAgents() ([]VerificationAgentEntry, error) {
	listedAgents, err := s.verificationAgentService.ListAgents()
	if err != nil {
		return nil, err
	}

	entries := make([]VerificationAgentEntry, 0, len(listedAgents))
	for _, agent := range listedAgents {
		entries = append(entries, VerificationAgentEntry{
			MaxCPU:            agent.MaxCPU,
			MaxRAMGb:          agent.MaxRAMGb,
			MaxDiskGb:         agent.MaxDiskGb,
			MaxConcurrentJobs: agent.MaxConcurrentJobs,
		})
	}

	return entries, nil
}

// attachBackupSizes records raw/compressed size from the size-of-record backup.
// Physical databases have no logical backup rows, so their size comes from the
// latest completed physical FULL backup; every other type uses the latest
// completed logical backup.
func (s *TelemetryService) attachBackupSizes(
	entry *DatabaseEntry,
	db *databases.Database,
) error {
	if db.Type == databases.DatabaseTypePostgresPhysical {
		return s.attachPhysicalBackupSizes(entry, db.ID)
	}

	backup, err := s.backupService.GetLatestCompletedBackup(db.ID)
	if err != nil {
		return err
	}

	if backup == nil {
		return nil
	}

	if backup.BackupSizeMb > 0 {
		entry.BackupSizeMb = int64(math.Ceil(backup.BackupSizeMb))
	}

	if backup.BackupRawDbSizeMb > 0 {
		entry.RawSizeMb = int64(math.Ceil(backup.BackupRawDbSizeMb))
	}

	return nil
}

func (s *TelemetryService) attachPhysicalBackupSizes(
	entry *DatabaseEntry,
	databaseID uuid.UUID,
) error {
	fullBackup, err := s.physicalBackupService.GetLatestCompletedFullBackup(databaseID)
	if err != nil {
		return err
	}

	if fullBackup == nil {
		return nil
	}

	if fullBackup.BackupSizeMb != nil && *fullBackup.BackupSizeMb > 0 {
		entry.BackupSizeMb = int64(math.Ceil(*fullBackup.BackupSizeMb))
	}

	if fullBackup.RawSizeMb != nil && *fullBackup.RawSizeMb > 0 {
		entry.RawSizeMb = int64(math.Ceil(*fullBackup.RawSizeMb))
	}

	return nil
}

// isDatabaseActive returns true when a database should be counted in telemetry.
//
//   - HealthStatus == AVAILABLE   → active.
//   - HealthStatus == UNAVAILABLE → not active (healthcheck is on and the DB is down).
//   - HealthStatus == nil         → healthcheck is disabled; active only if a
//     successful backup happened inside `since`.
func (s *TelemetryService) isDatabaseActive(
	db *databases.Database,
	since time.Time,
) (bool, error) {
	if db.HealthStatus != nil {
		return *db.HealthStatus == databases.HealthStatusAvailable, nil
	}

	return s.backupService.HasSuccessfulBackupSince(db.ID, since)
}

func buildDatabaseEntry(db *databases.Database) (DatabaseEntry, bool) {
	switch db.Type {
	case databases.DatabaseTypePostgresLogical:
		// The legacy POSTGRES type is the same engine now labelled POSTGRES_LOGICAL;
		// analytics counts the two as one.
		if db.PostgresqlLogical == nil {
			return DatabaseEntry{}, false
		}
		return DatabaseEntry{Type: string(db.Type), Version: string(db.PostgresqlLogical.Version)}, true
	case databases.DatabaseTypePostgresPhysical:
		if db.PostgresqlPhysical == nil {
			return DatabaseEntry{}, false
		}
		return DatabaseEntry{
			Type:       string(db.Type),
			Version:    string(db.PostgresqlPhysical.Version),
			BackupType: string(db.PostgresqlPhysical.BackupType),
		}, true
	case databases.DatabaseTypeMysql:
		if db.Mysql == nil {
			return DatabaseEntry{}, false
		}
		return DatabaseEntry{Type: string(db.Type), Version: string(db.Mysql.Version)}, true
	case databases.DatabaseTypeMariadb:
		if db.Mariadb == nil {
			return DatabaseEntry{}, false
		}
		return DatabaseEntry{Type: string(db.Type), Version: string(db.Mariadb.Version)}, true
	case databases.DatabaseTypeMongodb:
		if db.Mongodb == nil {
			return DatabaseEntry{}, false
		}
		return DatabaseEntry{Type: string(db.Type), Version: string(db.Mongodb.Version)}, true
	}

	return DatabaseEntry{}, false
}

func (s *TelemetryService) collectStorageTypes() ([]string, error) {
	allStorages, err := s.storageService.GetAllStorages()
	if err != nil {
		return nil, err
	}

	types := make([]string, 0, len(allStorages))
	for _, st := range allStorages {
		key := string(st.Type)
		if key == "" {
			continue
		}

		types = append(types, key)
	}

	sort.Strings(types)
	return types, nil
}

func (s *TelemetryService) collectNotifierTypes() ([]string, error) {
	allNotifiers, err := s.notifierService.GetAllNotifiers()
	if err != nil {
		return nil, err
	}

	types := make([]string, 0, len(allNotifiers))
	for _, n := range allNotifiers {
		key := string(n.NotifierType)
		if key == "" {
			continue
		}

		types = append(types, key)
	}

	sort.Strings(types)
	return types, nil
}

func capStrings(in []string) []string {
	if len(in) > maxArrayEntries {
		return in[:maxArrayEntries]
	}

	return in
}

func capDatabases(in []DatabaseEntry) []DatabaseEntry {
	if len(in) > maxArrayEntries {
		return in[:maxArrayEntries]
	}

	return in
}

func capAgents(in []VerificationAgentEntry) []VerificationAgentEntry {
	if len(in) > maxArrayEntries {
		return in[:maxArrayEntries]
	}

	return in
}