pangolin/server/internal/nodes/store.go

package nodes

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

	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
)

// NodeRow holds a node's essential fields from the nodes table.
type NodeRow struct {
	ID         int64
	UUID       string
	Status     string
	RealityPBK string
	RealitySNI string
	Endpoint   string
	Hy2Port    sql.NullInt32
}

// NodeStore is the persistence interface used by the nodes domain handlers.
// All methods are context-aware and safe for concurrent use.
type NodeStore interface {
	// NodeByUUID looks up a node record by its UUID.
	// Returns (nil, nil) when no matching row exists.
	NodeByUUID(ctx context.Context, uuid string) (*NodeRow, error)

	// ConfigVersion returns the current global directory version.
	// This is the version from the directory_version singleton table.
	ConfigVersion(ctx context.Context) (int64, error)

	// ActiveNodeUUIDs returns the UUIDs of all nodes with status 'up' or 'draining'.
	// Used by Broadcast to enumerate delivery targets.
	ActiveNodeUUIDs(ctx context.Context) ([]string, error)

	// CredentialsForNode returns the active data-plane credentials for nodeUUID.
	// Returns an empty slice until task 5d creates the connect_credentials table.
	CredentialsForNode(ctx context.Context, nodeUUID string) ([]*agentv1.Credential, error)

	// UserIDByDpUUID maps a data-plane UUID to the owning user's internal ID.
	// Returns (0, false, nil) if the dp_uuid is unknown or the user is inactive.
	UserIDByDpUUID(ctx context.Context, dpUUID string) (int64, bool, error)

	// AccumulateUsage adds bytes and minutes to usage_daily for userID on date.
	// Uses INSERT … ON DUPLICATE KEY UPDATE (idempotent within a day).
	AccumulateUsage(ctx context.Context, userID int64, date time.Time,
		bytesUp, bytesDown int64, minutes int64) error
}

// SQLNodeStore implements NodeStore against a MySQL 8.x database.
type SQLNodeStore struct {
	db *sql.DB
}

// NewSQLNodeStore creates a SQLNodeStore backed by db.
func NewSQLNodeStore(db *sql.DB) *SQLNodeStore {
	return &SQLNodeStore{db: db}
}

// NodeByUUID looks up a node by UUID. Returns (nil, nil) when not found.
func (s *SQLNodeStore) NodeByUUID(ctx context.Context, uuid string) (*NodeRow, error) {
	const q = `
		SELECT id, uuid, status, reality_pbk, reality_sni, endpoint, hy2_port
		FROM nodes
		WHERE uuid = ?
	`
	var n NodeRow
	err := s.db.QueryRowContext(ctx, q, uuid).Scan(
		&n.ID, &n.UUID, &n.Status,
		&n.RealityPBK, &n.RealitySNI, &n.Endpoint, &n.Hy2Port,
	)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("nodes.SQLNodeStore.NodeByUUID: %w", err)
	}
	return &n, nil
}

// ConfigVersion returns the current global directory version.
// Returns 0 with nil error if the directory_version row does not yet exist.
func (s *SQLNodeStore) ConfigVersion(ctx context.Context) (int64, error) {
	var v int64
	err := s.db.QueryRowContext(ctx,
		`SELECT version FROM directory_version WHERE id = 1`).Scan(&v)
	if err == sql.ErrNoRows {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("nodes.SQLNodeStore.ConfigVersion: %w", err)
	}
	return v, nil
}

// ActiveNodeUUIDs returns UUIDs of all nodes with status 'up' or 'draining'.
func (s *SQLNodeStore) ActiveNodeUUIDs(ctx context.Context) ([]string, error) {
	rows, err := s.db.QueryContext(ctx,
		`SELECT uuid FROM nodes WHERE status IN ('up', 'draining')`)
	if err != nil {
		return nil, fmt.Errorf("nodes.SQLNodeStore.ActiveNodeUUIDs: %w", err)
	}
	defer rows.Close()

	var uuids []string
	for rows.Next() {
		var u string
		if err := rows.Scan(&u); err != nil {
			return nil, fmt.Errorf("nodes.SQLNodeStore.ActiveNodeUUIDs scan: %w", err)
		}
		uuids = append(uuids, u)
	}
	return uuids, rows.Err()
}

// CredentialsForNode returns active credentials for nodeUUID.
// Stub: returns empty slice until task 5d creates the connect_credentials table.
func (s *SQLNodeStore) CredentialsForNode(_ context.Context, _ string) ([]*agentv1.Credential, error) {
	return nil, nil
}

// UserIDByDpUUID resolves a data-plane UUID to an active user's internal ID.
func (s *SQLNodeStore) UserIDByDpUUID(ctx context.Context, dpUUID string) (int64, bool, error) {
	var userID int64
	err := s.db.QueryRowContext(ctx,
		`SELECT id FROM users WHERE dp_uuid = ? AND status = 'active' LIMIT 1`, dpUUID,
	).Scan(&userID)
	if err == sql.ErrNoRows {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, fmt.Errorf("nodes.SQLNodeStore.UserIDByDpUUID: %w", err)
	}
	return userID, true, nil
}

// AccumulateUsage adds bytes/minutes to usage_daily for the given user and date.
func (s *SQLNodeStore) AccumulateUsage(
	ctx context.Context, userID int64, date time.Time,
	bytesUp, bytesDown int64, minutes int64,
) error {
	const q = `
		INSERT INTO usage_daily (user_id, date, bytes_up, bytes_down, minutes_used)
		VALUES (?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE
			bytes_up     = bytes_up     + VALUES(bytes_up),
			bytes_down   = bytes_down   + VALUES(bytes_down),
			minutes_used = minutes_used + VALUES(minutes_used)
	`
	if _, err := s.db.ExecContext(ctx, q,
		userID, date.Format("2006-01-02"), bytesUp, bytesDown, minutes,
	); err != nil {
		return fmt.Errorf("nodes.SQLNodeStore.AccumulateUsage: %w", err)
	}
	return nil
}