johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bidder

import (
	"cmp"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/url"
	"slices"
	"strconv"
	"strings"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	contract "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/guestnet"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/guestseed"
)

const abandonTimeout = 30 * time.Second

type provisionRequest struct {
	rfpURI string

	rfpCID string

	vmURI string

	vmCID string

	acceptURI string

	acceptCID string

	bidURI string

	bidCID string

	requesterDID string

	acceptedAt string

	vmPayload market.ResolvedRecord

	bidConfig market.StrongRef

	bidConfigValue map[string]any

	owner *metav1.OwnerReference
}

type vmRecord struct {
	Kind string `json:"kind"`

	CPUs int32 `json:"cpus"`

	Mem string `json:"mem"`

	Disk string `json:"disk"`

	Role string `json:"role"`

	UserData string `json:"user_data"`
}

func parseVMRecord(value map[string]any) (vmRecord, error) {
	var out vmRecord
	raw, err := json.Marshal(value)
	if err != nil {
		return out, err
	}
	if err := json.Unmarshal(raw, &out); err != nil {
		return out, fmt.Errorf("bidder: the RFP payload is not a compute.vm record: %w", err)
	}
	out.CPUs = cmp.Or(out.CPUs, 2)
	if strings.TrimSpace(out.Mem) == "" {
		out.Mem = "4G"
	}
	switch {
	case out.Role == "":
		return out, errors.New("bidder: the RFP payload names no vm.role")
	case out.CPUs < 1:
		return out, fmt.Errorf("bidder: the RFP payload names %d cpus", out.CPUs)
	case strings.TrimSpace(out.UserData) == "":
		return out, errors.New("bidder: the RFP payload carries no user_data; a guest without it is unreachable")
	}
	return out, nil
}

func (b *Bidder) owns(handle string) bool {
	return handle == b.cfg.ProviderHandle
}

func (b *Bidder) provision(ctx context.Context, req provisionRequest) (string, error) {
	if b.vault == nil {
		return "", errors.New("bidder: the compute provider has no guest vault")
	}
	if b.cfg.GuestURL == "" {
		return "", errors.New("bidder: the compute provider has no guest identity issuer URL")
	}
	record, err := parseVMRecord(req.vmPayload.Value)
	if err != nil {
		return "", err
	}
	if record.Kind == string(v1alpha1.ComputeKindDeno) {
		return b.provisionSandbox(ctx, req, record)
	}
	acceptKey := b.guests.AcceptKey(req.acceptURI, req.acceptCID)
	if acceptKey != "" {
		reuse, ok := b.vms.Find(func(vm v1alpha1.ComputeVM) bool {
			return b.owns(vm.Spec.Owner.Handle) && b.guests.AcceptKeyOf(&vm) == acceptKey && vm.DeletionTimestamp == nil &&
				!b.guests.Releasing(&vm) && !b.guests.Ended(&vm)
		})
		if ok {
			b.log.Info("a guest ComputeVM is reused for the accepted contract", "computevm", reuse.Name, "accept", req.acceptURI)
			return reuse.Name, nil
		}
	}
	if err := b.checkCapacity(); err != nil {
		return "", err
	}
	name, err := vmNameForAccept(req.requesterDID, acceptKey)
	if err != nil {
		return "", err
	}
	secret, secretHash, err := b.guests.LoginSecret(acceptKey)
	if err != nil {
		return "", err
	}
	issuer := strings.TrimRight(b.cfg.GuestURL, "/")
	bundle := map[string]any{
		"$type":               market.AcceptNSID,
		"accept":              market.NewStrongRef(req.acceptURI, req.acceptCID).Map(),
		"rfp":                 market.NewStrongRef(req.rfpURI, req.rfpCID).Map(),
		"bid":                 nil,
		"bid_config":          map[string]any{"uri": req.bidConfig.URI, "cid": req.bidConfig.CID, "value": req.bidConfigValue},
		"vm":                  map[string]any{"uri": req.vmURI, "cid": req.vmCID, "value": req.vmPayload.Value},
		"guest_onnetwork_url": issuer + guestnet.EventsPath,
	}
	if req.bidURI != "" {
		bundle["bid"] = market.NewStrongRef(req.bidURI, req.bidCID).Map()
	}
	withBundle, err := guestseed.InjectAcceptBundle(record.UserData, bundle, guestseed.AcceptPathVM)
	if err != nil {
		return "", err
	}
	webrtc, err := b.webrtcPort(name, withBundle)
	if err != nil {
		return "", err
	}
	hostURL, hostPort := b.GuestHostURL()
	userData, err := guestseed.Provision(guestseed.ProvisionOptions{
		UserData: withBundle,
		Login: &guestseed.GuestLoginOptions{
			URL:        issuer + guestseed.GuestLoginRoute,
			Role:       name,
			BindSecret: secret,
			Mount:      b.cfg.GuestLoginMount,
		},
		AcceptPath:     guestseed.AcceptPathVM,
		FallbackURL:    b.cfg.GuestURL,
		BearerAudience: b.guests.Audience(name),
		Reporter:       true,
		BaseURL:        true,
		HostAddress:    v1alpha1.GuestHostAddress,
		TLSPort:        b.cfg.GuestTLSPort,
		HostURL:        hostURL,
		Downloads:      b.cfg.GuestDownloads,
		WebRTC:         webrtc,
	})
	if err != nil {
		return "", err
	}
	path, entry := b.guests.UserData(name, userData)
	if err := b.vault.Write(ctx, b.cfg.GuestVaultMount, path, entry); err != nil {
		return "", fmt.Errorf("bidder: writing the guest user data: %w", err)
	}
	vm, err := b.guests.ComputeVM(GuestVM{
		Name:             name,
		Handle:           b.cfg.ProviderHandle,
		Shape:            v1alpha1.ComputeVMShape{Cpus: record.CPUs, Mem: record.Mem, Disk: record.Disk, Role: record.Role},
		RequesterDID:     req.requesterDID,
		VaultServer:      b.cfg.GuestVaultServer,
		VaultMount:       b.cfg.GuestVaultMount,
		StopGraceSeconds: b.cfg.StopGraceSeconds,
		Login:            &v1alpha1.ComputeVMGuestLogin{Mount: b.cfg.GuestLoginMount, BindSecretSha256: secretHash},
		Accept:           commonv1.StrongRef{URI: req.acceptURI, CID: req.acceptCID},
		HostPorts:        b.guestHostPorts(userData, hostPort),
		LANUDP:           b.cfg.GuestLANUDP,
		HostUDP:          b.cfg.GuestHostUDP,
		WebRTCUDPPort:    webrtcUDPPort(webrtc),
		Owner:            req.owner,
	})
	if err != nil {
		return "", err
	}
	vm.Spec.AssociationExpiresAt = b.associationExpiry(ctx, req.requesterDID, req.acceptedAt)
	created := true
	if err := b.intents.CreateComputeVM(ctx, *vm); err != nil {
		if !errors.Is(err, contract.ErrExists) {
			return "", fmt.Errorf("bidder: creating the ComputeVM: %w", err)
		}
		existing, found := b.awaitVM(ctx, name)
		if !found || b.guests.AcceptKeyOf(&existing) != acceptKey || !b.owns(existing.Spec.Owner.Handle) {
			return "", fmt.Errorf("bidder: creating the ComputeVM: %w", err)
		}
		created = false
	}
	if err := b.registerGrant(ctx, name, req.requesterDID, record.Role); err != nil {
		if created {
			b.abandon(ctx, name)
		}
		return "", err
	}
	b.log.Info("a guest ComputeVM is applied", "computevm", name, "cpus", record.CPUs, "role", record.Role, "requester", req.requesterDID)
	return name, nil
}

func (b *Bidder) webrtcPort(name, userData string) (*guestseed.WebRTCOptions, error) {
	if b.webrtcPorts == nil || b.cfg.GuestLANAddresses == nil || !guestseed.GuestBinariesNeeded(userData).WebRTC {
		return nil, nil
	}
	addresses := b.cfg.GuestLANAddresses()
	if len(addresses) == 0 {
		b.log.Info("the bidder host has no LAN address; the guest's webrtc gets no forwarded LAN UDP port", "computevm", name)
		return nil, nil
	}
	if b.synced.Load() {
		live := map[string]bool{name: true}
		for _, vm := range b.vms.Filter(func(v1alpha1.ComputeVM) bool { return true }) {
			live[vm.Name] = true
		}
		for _, held := range b.webrtcPorts.Held() {
			if !live[held] {
				if err := b.webrtcPorts.Release(held); err != nil {
					b.log.Warn("releasing the webrtc UDP port of a gone ComputeVM failed", "computevm", held, "err", err)
				}
			}
		}
	}
	port, err := b.webrtcPorts.Allocate(name)
	if err != nil {
		return nil, fmt.Errorf("bidder: a webrtc UDP port for %s: %w", name, err)
	}
	if err := guestnet.ValidWebRTCUDPPort(port); err != nil {
		_ = b.webrtcPorts.Release(name)
		return nil, fmt.Errorf("bidder: %w", err)
	}
	b.log.Info("the guest's webrtc gets a forwarded LAN UDP port", "computevm", name, "port", port, "lanAddresses", len(addresses))
	return &guestseed.WebRTCOptions{UDPPort: port, HostAddresses: addresses}, nil
}

func webrtcUDPPort(w *guestseed.WebRTCOptions) int32 {
	if w == nil {
		return 0
	}
	return int32(w.UDPPort)
}

func (b *Bidder) GuestHostURL() (string, int) {
	if b.cfg.GuestLocalAddress == nil {
		return "", 0
	}
	raw := strings.TrimSpace(b.cfg.GuestLocalAddress())
	if raw == "" {
		return "", 0
	}
	u, err := url.Parse(raw)
	if err != nil || u.Scheme != "http" {
		return "", 0
	}
	port, err := strconv.Atoi(u.Port())
	if err != nil || port < 1 || port > 65535 {
		return "", 0
	}
	return "http://" + net.JoinHostPort(guestnet.HostAddress, strconv.Itoa(port)), port
}

func (b *Bidder) awaitVM(ctx context.Context, name string) (v1alpha1.ComputeVM, bool) {
	waitCtx, cancel := context.WithTimeout(ctx, AwaitTimeout)
	defer cancel()
	var vm v1alpha1.ComputeVM
	found := false
	_ = b.vms.Await(waitCtx, func() bool {
		vm, found = b.vms.Get(name)
		return found
	})
	return vm, found
}

func (b *Bidder) checkCapacity() error {
	limit := b.cfg.MaxConcurrentGuests
	if limit <= 0 {
		return nil
	}
	live := len(b.vms.Filter(func(vm v1alpha1.ComputeVM) bool {
		return vm.DeletionTimestamp == nil && vm.Spec.Owner.Handle == b.cfg.ProviderHandle && !b.guests.Ended(&vm) && !b.guests.Releasing(&vm)
	}))
	if int32(live) >= limit {
		return fmt.Errorf("bidder: the %d guest quota is full", limit)
	}
	return nil
}

func (b *Bidder) registerGrant(ctx context.Context, name, requesterDID, role string) error {
	grant, err := b.guests.Grant(b.cfg.DID, requesterDID, role, name)
	if err != nil {
		return err
	}
	if err := b.intents.CreateGuestIdentityGrant(ctx, grant); err != nil && !errors.Is(err, contract.ErrExists) {
		return fmt.Errorf("bidder: writing the guest identity grant: %w", err)
	}
	return nil
}

func (b *Bidder) abandon(ctx context.Context, name string) {
	cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), abandonTimeout)
	defer cancel()
	if err := b.intents.DeleteComputeVM(cleanup, name); err != nil {
		b.log.Warn("releasing an unprovisioned ComputeVM failed", "computevm", name, "err", err)
	}
}

func vmNameForAccept(requesterDID, acceptKey string) (string, error) {
	prefix := vmNamePrefix(requesterDID)
	if acceptKey == "" {
		buf := make([]byte, 4)
		if _, err := rand.Read(buf); err != nil {
			return "", fmt.Errorf("bidder: drawing a ComputeVM name: %w", err)
		}
		return prefix + "-" + hex.EncodeToString(buf), nil
	}
	sum := sha256.Sum256([]byte(acceptKey))
	return prefix + "-" + hex.EncodeToString(sum[:4]), nil
}

func vmNamePrefix(requesterDID string) string {
	parts := strings.Split(requesterDID, ":")
	short := strings.ToLower(parts[len(parts)-1])
	var clean strings.Builder
	for _, r := range short {
		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '-' || r == '_' || r == '.' {
			clean.WriteRune(r)
		}
	}
	text := strings.Trim(clean.String(), "-._")
	if len(text) > 12 {
		text = text[:12]
	}
	return "fc-" + cmp.Or(text, "guest")
}

func (b *Bidder) guestHostPorts(userData string, hostPort int) []int32 {
	ports := guestnet.HostPorts(append(append(slices.Clone(b.cfg.GuestHostPorts), hostPort), guestseed.HostPortsOf(userData, b.cfg.GuestTLSPort)...))
	out := make([]int32, 0, len(ports))
	for _, p := range ports {
		out = append(out, int32(p))
	}
	return out
}