pulse/pkg/pbs/client.go

package pbs

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"
	
	"github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil"
	"github.com/rs/zerolog/log"
)

// Client represents a Proxmox Backup Server API client
type Client struct {
	baseURL    string
	httpClient *http.Client
	auth       auth
	config     ClientConfig
}

// ClientConfig holds configuration for the PBS client
type ClientConfig struct {
	Host        string
	User        string
	Password    string
	TokenName   string
	TokenValue  string
	Fingerprint string
	VerifySSL   bool
	Timeout     time.Duration
}

// auth represents authentication details
type auth struct {
	user       string
	realm      string
	ticket     string
	csrfToken  string
	tokenName  string
	tokenValue string
	expiresAt  time.Time
}

// NewClient creates a new PBS API client
func NewClient(cfg ClientConfig) (*Client, error) {
	var user, realm string
	
	// For token auth, user might be empty or in a different format
	if cfg.TokenName != "" && cfg.TokenValue != "" {
		// Token authentication - parse the token name to extract user info if needed
		if strings.Contains(cfg.TokenName, "!") {
			// Token name contains full format: user@realm!tokenname
			parts := strings.Split(cfg.TokenName, "!")
			if len(parts) == 2 && strings.Contains(parts[0], "@") {
				userParts := strings.Split(parts[0], "@")
				if len(userParts) == 2 {
					user = userParts[0]
					realm = userParts[1]
					// Update token name to just the token part
					cfg.TokenName = parts[1]
				}
			}
		} else if cfg.User != "" {
			// User provided separately
			parts := strings.Split(cfg.User, "@")
			if len(parts) == 2 {
				user = parts[0]
				realm = parts[1]
			} else {
				// If no realm specified, default to pbs
				user = cfg.User
				realm = "pbs"
			}
		} else {
			return nil, fmt.Errorf("token authentication requires user information either in token name (user@realm!tokenname) or user field")
		}
	} else {
		// Password authentication - user@realm format is required
		parts := strings.Split(cfg.User, "@")
		if len(parts) != 2 {
			return nil, fmt.Errorf("invalid user format, expected user@realm")
		}
		user = parts[0]
		realm = parts[1]
	}

	// Create HTTP client with proper TLS configuration
	httpClient := tlsutil.CreateHTTPClient(cfg.VerifySSL, cfg.Fingerprint)
	// Override timeout if specified
	if cfg.Timeout > 0 {
		httpClient.Timeout = cfg.Timeout
	}

	client := &Client{
		baseURL:    strings.TrimSuffix(cfg.Host, "/") + "/api2/json",
		httpClient: httpClient,
		config:     cfg,
		auth: auth{
			user:       user,
			realm:      realm,
			tokenName:  cfg.TokenName,
			tokenValue: cfg.TokenValue,
		},
	}

	// Authenticate if using password
	if cfg.Password != "" && cfg.TokenName == "" {
		if err := client.authenticate(context.Background()); err != nil {
			return nil, fmt.Errorf("authentication failed: %w", err)
		}
	}

	return client, nil
}

// authenticate performs password-based authentication
func (c *Client) authenticate(ctx context.Context) error {
	data := url.Values{
		"username": {c.auth.user + "@" + c.auth.realm},
		"password": {c.config.Password},
	}

	req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/access/ticket", strings.NewReader(data.Encode()))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		if resp.StatusCode == 401 || resp.StatusCode == 403 {
			return fmt.Errorf("authentication failed (status %d): %s", resp.StatusCode, string(body))
		}
		return fmt.Errorf("authentication failed: %s", string(body))
	}

	var result struct {
		Data struct {
			Ticket            string `json:"ticket"`
			CSRFPreventionToken string `json:"CSRFPreventionToken"`
		} `json:"data"`
	}

	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return err
	}

	c.auth.ticket = result.Data.Ticket
	c.auth.csrfToken = result.Data.CSRFPreventionToken
	c.auth.expiresAt = time.Now().Add(2 * time.Hour) // PBS tickets expire after 2 hours

	return nil
}

// request performs an API request
func (c *Client) request(ctx context.Context, method, path string, data url.Values) (*http.Response, error) {
	// Re-authenticate if needed
	if c.config.Password != "" && c.auth.tokenName == "" && time.Now().After(c.auth.expiresAt) {
		if err := c.authenticate(ctx); err != nil {
			return nil, fmt.Errorf("re-authentication failed: %w", err)
		}
	}

	var body io.Reader
	if data != nil {
		body = strings.NewReader(data.Encode())
	}

	req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, body)
	if err != nil {
		return nil, err
	}

	// Set headers
	if data != nil {
		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	}

	// Set authentication
	if c.auth.tokenName != "" && c.auth.tokenValue != "" {
		// API token authentication
		req.Header.Set("Authorization", fmt.Sprintf("PBSAPIToken=%s@%s!%s:%s",
			c.auth.user, c.auth.realm, c.auth.tokenName, c.auth.tokenValue))
	} else if c.auth.ticket != "" {
		// Ticket authentication
		req.Header.Set("Cookie", "PBSAuthCookie="+c.auth.ticket)
		if method != "GET" && c.auth.csrfToken != "" {
			req.Header.Set("CSRFPreventionToken", c.auth.csrfToken)
		}
	}

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}

	// Check for errors
	if resp.StatusCode >= 400 {
		defer resp.Body.Close()
		body, _ := io.ReadAll(resp.Body)
		
		// Create base error
		err := fmt.Errorf("API error %d: %s", resp.StatusCode, string(body))
		
		// Wrap with appropriate error type
		if resp.StatusCode == 401 || resp.StatusCode == 403 {
			return nil, fmt.Errorf("authentication error: %w", err)
		}
		
		return nil, err
	}

	return resp, nil
}

// get performs a GET request
func (c *Client) get(ctx context.Context, path string) (*http.Response, error) {
	return c.request(ctx, "GET", path, nil)
}

// Version represents PBS version information
type Version struct {
	Version string `json:"version"`
	Release string `json:"release"`
	Repoid  string `json:"repoid"`
}

// Datastore represents a PBS datastore
type Datastore struct {
	Store     string `json:"store"`
	Total     int64  `json:"total"`
	Used      int64  `json:"used"`
	Avail     int64  `json:"avail"`
	GCStatus  string `json:"gc-status,omitempty"`
	Error     string `json:"error,omitempty"`
}

// GetVersion returns PBS version information
func (c *Client) GetVersion(ctx context.Context) (*Version, error) {
	resp, err := c.get(ctx, "/version")
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	var result struct {
		Data Version `json:"data"`
	}

	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}

	return &result.Data, nil
}

// GetDatastores returns all datastores
func (c *Client) GetDatastores(ctx context.Context) ([]Datastore, error) {
	resp, err := c.get(ctx, "/admin/datastore")
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	var result struct {
		Data []Datastore `json:"data"`
	}

	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}

	return result.Data, nil
}

// Namespace represents a PBS namespace
type Namespace struct {
	NS     string `json:"ns"`
	Path   string `json:"path"`
	Name   string `json:"name"`
	Parent string `json:"parent,omitempty"`
}

// BackupGroup represents a group of backups for a specific VM/CT
type BackupGroup struct {
	BackupType  string   `json:"backup-type"`  // "vm" or "ct"
	BackupID    string   `json:"backup-id"`    // VMID
	LastBackup  int64    `json:"last-backup"`  // Unix timestamp
	BackupCount int      `json:"backup-count"`
	Files       []string `json:"files,omitempty"`
	Owner       string   `json:"owner,omitempty"`
}

// BackupSnapshot represents a single backup snapshot
type BackupSnapshot struct {
	BackupType   string         `json:"backup-type"`  // "vm" or "ct"
	BackupID     string         `json:"backup-id"`    // VMID
	BackupTime   int64          `json:"backup-time"`  // Unix timestamp
	Files        []interface{}  `json:"files,omitempty"` // Can be strings or objects
	Size         int64          `json:"size"`
	Protected    bool           `json:"protected"`
	Comment      string         `json:"comment,omitempty"`
	Owner        string         `json:"owner,omitempty"`
	Verification interface{}    `json:"verification,omitempty"` // Can be string or object
}

// ListNamespaces lists namespaces for a datastore
func (c *Client) ListNamespaces(ctx context.Context, datastore string, parentNamespace string, maxDepth int) ([]Namespace, error) {
	path := fmt.Sprintf("/admin/datastore/%s/namespace", datastore)
	
	// Build query parameters
	params := url.Values{}
	if parentNamespace != "" {
		params.Set("ns", parentNamespace)
	}
	if maxDepth > 0 {
		params.Set("max-depth", fmt.Sprintf("%d", maxDepth))
	}
	
	if len(params) > 0 {
		path += "?" + params.Encode()
	}
	
	resp, err := c.get(ctx, path)
	if err != nil {
		// If namespace endpoint doesn't exist (older PBS versions), return empty list
		if strings.Contains(err.Error(), "404") {
			return []Namespace{}, nil
		}
		return nil, err
	}
	defer resp.Body.Close()

	var result struct {
		Data []Namespace `json:"data"`
	}

	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}

	return result.Data, nil
}
// ListBackupGroups lists all backup groups in a datastore/namespace
func (c *Client) ListBackupGroups(ctx context.Context, datastore string, namespace string) ([]BackupGroup, error) {
	path := fmt.Sprintf("/admin/datastore/%s/groups", datastore)
	
	// Add namespace parameter if provided
	params := url.Values{}
	if namespace != "" {
		params.Set("ns", namespace)
	}
	
	if len(params) > 0 {
		path = path + "?" + params.Encode()
	}
	
	// Log the API call
	log.Debug().Str("url", c.baseURL+path).Msg("PBS API: ListBackupGroups")
	
	resp, err := c.get(ctx, path)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
	}

	var result struct {
		Data []BackupGroup `json:"data"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, fmt.Errorf("failed to decode backup groups: %w", err)
	}

	log.Debug().
		Str("namespace", namespace).
		Int("count", len(result.Data)).
		Msg("PBS API: Backup groups found")
	return result.Data, nil
}

// ListBackupSnapshots lists all snapshots for a specific backup group
func (c *Client) ListBackupSnapshots(ctx context.Context, datastore string, namespace string, backupType string, backupID string) ([]BackupSnapshot, error) {
	path := fmt.Sprintf("/admin/datastore/%s/snapshots", datastore)
	
	// Build parameters
	params := url.Values{}
	if namespace != "" {
		params.Set("ns", namespace)
	}
	params.Set("backup-type", backupType)
	params.Set("backup-id", backupID)
	
	path = path + "?" + params.Encode()
	
	resp, err := c.get(ctx, path)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body))
	}

	var result struct {
		Data []BackupSnapshot `json:"data"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, fmt.Errorf("failed to decode snapshots: %w", err)
	}

	return result.Data, nil
}

// ListAllBackups fetches all backups from all namespaces concurrently
func (c *Client) ListAllBackups(ctx context.Context, datastore string, namespaces []string) (map[string][]BackupSnapshot, error) {
	type namespaceResult struct {
		namespace string
		snapshots []BackupSnapshot
		err      error
	}
	
	// Channel for results
	resultCh := make(chan namespaceResult, len(namespaces))
	
	// WaitGroup to track goroutines
	var wg sync.WaitGroup
	
	// Semaphore to limit concurrent requests
	sem := make(chan struct{}, 3) // Max 3 concurrent requests
	
	// Fetch backups from each namespace concurrently
	for _, ns := range namespaces {
		wg.Add(1)
		go func(namespace string) {
			defer wg.Done()
			
			// Acquire semaphore
			sem <- struct{}{}
			defer func() { <-sem }()
			
			// Get groups first
			groups, err := c.ListBackupGroups(ctx, datastore, namespace)
			if err != nil {
				log.Error().
					Str("datastore", datastore).
					Str("namespace", namespace).
					Err(err).
					Msg("Failed to list backup groups")
				resultCh <- namespaceResult{namespace: namespace, err: err}
				return
			}
			
			log.Info().
				Str("datastore", datastore).
				Str("namespace", namespace).
				Int("groups", len(groups)).
				Msg("Found backup groups")
			
			var allSnapshots []BackupSnapshot
			
			// For each group, get snapshots
			for _, group := range groups {
				snapshots, err := c.ListBackupSnapshots(ctx, datastore, namespace, group.BackupType, group.BackupID)
				if err != nil {
					log.Error().
						Str("datastore", datastore).
						Str("namespace", namespace).
						Str("type", group.BackupType).
						Str("id", group.BackupID).
						Err(err).
						Msg("Failed to list snapshots")
					continue
				}
				allSnapshots = append(allSnapshots, snapshots...)
			}
			
			resultCh <- namespaceResult{
				namespace: namespace,
				snapshots: allSnapshots,
				err:      nil,
			}
		}(ns)
	}
	
	// Close channel when all goroutines complete
	go func() {
		wg.Wait()
		close(resultCh)
	}()
	
	// Collect results
	results := make(map[string][]BackupSnapshot)
	var errors []error
	
	for result := range resultCh {
		if result.err != nil {
			errors = append(errors, fmt.Errorf("namespace %s: %w", result.namespace, result.err))
		} else {
			results[result.namespace] = result.snapshots
		}
	}
	
	// Return combined error if any occurred
	if len(errors) > 0 {
		return results, fmt.Errorf("errors fetching backups: %v", errors)
	}
	
	return results, nil
}