pmg/internal/audit/cloud_client.go

package audit

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/safedep/dry/cloud"
	"github.com/safedep/dry/cloud/endpointsync"
	"github.com/safedep/dry/log"
	"github.com/safedep/pmg/config"
	"github.com/safedep/pmg/internal/cloudauth"
	appVersion "github.com/safedep/pmg/internal/version"
)

// ErrSyncInProgress is returned by DrainToCloud when another process already
// holds the cloud sync lock.
var ErrSyncInProgress = errors.New("another cloud sync is already in progress")

// DrainToCloud acquires the cross-process sync lock and drains pending audit
// events from the WAL to SafeDep Cloud, returning the number synced. It records
// the attempt (so a failing endpoint does not make every run retry). Callers
// are responsible for gating on cloud.enabled and for surfacing errors;
// ErrSyncInProgress is returned when the lock is held elsewhere.
//
// lockTimeout bounds acquiring the shared lock; syncTimeout bounds the drain.
func DrainToCloud(ctx context.Context, cfg *config.RuntimeConfig, lockTimeout, syncTimeout time.Duration) (int, error) {
	lock := NewSyncLock(cfg.CloudSyncLockPath())

	lockCtx, lockCancel := context.WithTimeout(ctx, lockTimeout)
	defer lockCancel()

	locked, err := lock.TryLockContext(lockCtx, 250*time.Millisecond)
	if err != nil {
		return 0, fmt.Errorf("acquire cloud sync lock: %w", err)
	}
	if !locked {
		return 0, ErrSyncInProgress
	}
	defer func() {
		if uerr := lock.Unlock(); uerr != nil {
			log.Warnf("failed to release cloud sync lock: %v", uerr)
		}
	}()

	bundle, err := NewSyncClientBundle(cfg)
	if err != nil {
		return 0, fmt.Errorf("init cloud sync client: %w", err)
	}
	defer func() {
		if cerr := bundle.Close(); cerr != nil {
			log.Warnf("failed to close cloud sync client: %v", cerr)
		}
	}()

	syncCtx, cancel := context.WithTimeout(ctx, syncTimeout)
	defer cancel()

	synced, syncErr := bundle.Sync(syncCtx)

	// Record the attempt on every outcome so a stuck endpoint does not make the
	// background auto-sync refire on the next invocation.
	if werr := WriteLastSyncAttempt(cfg.CloudSyncLastRunPath()); werr != nil {
		log.Warnf("failed to update cloud sync lastrun: %v", werr)
	}

	return synced, syncErr
}

// SyncClientBundle holds a SyncClient and its underlying cloud client.
// Callers must call Close() when done.
type SyncClientBundle struct {
	syncClient  *endpointsync.SyncClient
	cloudClient *cloud.Client
}

// Sync delivers pending events from the WAL to SafeDep Cloud.
func (b *SyncClientBundle) Sync(ctx context.Context) (int, error) {
	return b.syncClient.Sync(ctx)
}

func (b *SyncClientBundle) Close() error {
	var errs []error
	if b.syncClient != nil {
		if err := b.syncClient.Close(); err != nil {
			errs = append(errs, err)
		}
	}
	if b.cloudClient != nil {
		if err := b.cloudClient.Close(); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// NewSyncClientBundle creates an authenticated SyncClient connected to SafeDep Cloud.
func NewSyncClientBundle(cfg *config.RuntimeConfig) (*SyncClientBundle, error) {
	// Resolve credentials via keychain-first, env fallback chain. The keychain
	// resolver can be closed immediately because the data plane client extracts
	// the credential values when it is created.
	creds, closeResolver, err := cloudauth.ResolveCredentials()
	if err != nil {
		return nil, err
	}
	defer func() {
		if closeErr := closeResolver(); closeErr != nil {
			log.Warnf("failed to close keychain resolver: %v", closeErr)
		}
	}()

	cloudClient, err := cloud.NewDataPlaneClient("pmg", creds)
	if err != nil {
		return nil, fmt.Errorf("failed to create data plane client: %w", err)
	}

	transport := endpointsync.NewGrpcTransport(cloudClient.Connection())

	var identityOpts []endpointsync.EndpointIdentityOption
	if cfg.Config.Cloud.EndpointID != "" {
		identityOpts = append(identityOpts, endpointsync.WithEndpointID(cfg.Config.Cloud.EndpointID))
	}

	identity := endpointsync.NewEndpointIdentityResolver(identityOpts...)

	syncClient, err := endpointsync.NewSyncClient("pmg", pmgToolVersion(), transport, identity,
		endpointsync.WithWALPath(cfg.CloudSyncDBPath()))
	if err != nil {
		if closeErr := cloudClient.Close(); closeErr != nil {
			log.Warnf("failed to close cloud client after sync client init failure: %v", closeErr)
		}
		return nil, fmt.Errorf("failed to create sync client: %w", err)
	}

	return &SyncClientBundle{
		syncClient:  syncClient,
		cloudClient: cloudClient,
	}, nil
}

// pmgToolVersion returns the tool version stamped on cloud events, defaulting
// to "dev" for local builds.
func pmgToolVersion() string {
	if appVersion.Version == "" {
		return "dev"
	}
	return appVersion.Version
}