johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package atproto

import (
	"cmp"
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauthclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/pdsclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/ttlcache"
	recordclient "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/atproto"
)

const (
	ComponentKind = "Pds"

	ComponentName = "pds"

	ConditionReady = "Ready"

	ReasonRevoked = "Revoked"

	ReasonWaitingForCredential = "WaitingForCredential"

	ReasonVaultUnavailable = "VaultUnavailable"

	requestTimeout = 5 * time.Second

	minRefreshAfter = time.Second
)

type EndpointFunc func(workspace, kind, name string) (string, bool)

type VaultFunc func(workspace string) SecretStore

type TLSFunc func(ctx context.Context, workspace string, id *v1alpha1.MarketIdentity) (*tls.Config, error)

type Config struct {
	Endpoints EndpointFunc

	Vault VaultFunc

	Resolver *pdsclient.Resolver

	HTTPClient *http.Client

	TLS TLSFunc

	Sessions SessionsFunc

	OAuth OAuthOptions

	PLCURL string

	RecordHTTP *http.Client

	Log *slog.Logger
}

type Result struct {
	Status v1alpha1.MarketIdentityStatus

	RequeueAfter time.Duration
}

type Accounts struct {
	endpoints EndpointFunc

	vault VaultFunc

	log *slog.Logger

	http *http.Client

	resolver *pdsclient.Resolver

	tls TLSFunc

	sessions SessionsFunc

	oauth OAuthOptions

	states oauthclient.StateStore

	qr *ttlcache.Cache[string, *qrTransfer]

	browser *ttlcache.Cache[string, browserIssue]

	held heldAccounts

	records *recordclient.Client

	refs *ttlcache.Cache[refKey, refVerdict]

	verified *ttlcache.Cache[refKey, refVerdict]
}

func New(cfg Config) *Accounts {
	log := cfg.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	client := cfg.HTTPClient
	if client == nil {
		client = &http.Client{Timeout: requestTimeout}
	}
	resolver := cfg.Resolver
	if resolver == nil {
		resolver = &pdsclient.Resolver{HTTPClient: client, Timeout: requestTimeout}
	}
	states := cfg.OAuth.States
	if states == nil {
		states = &oauthclient.MemoryStateStore{}
	}
	records := recordclient.NewClient()
	records.HTTP = cmp.Or(cfg.RecordHTTP, records.HTTP)
	records.PLCURL = cmp.Or(strings.TrimSpace(cfg.PLCURL), records.PLCURL)
	return &Accounts{endpoints: cfg.Endpoints, vault: cfg.Vault, log: log, http: client, resolver: resolver, tls: cfg.TLS, sessions: cfg.Sessions, oauth: cfg.OAuth,
		states: states, qr: ttlcache.New[string, *qrTransfer](ttlcache.Options{}), browser: ttlcache.New[string, browserIssue](ttlcache.Options{}),
		records: records, refs: newRefCache(), verified: newVerifiedCache()}
}

func (a *Accounts) reconcile(ctx context.Context, workspace string, id *v1alpha1.MarketIdentity, status v1alpha1.MarketIdentityStatus) (Result, error) {
	generation := id.Generation
	if id.Spec.Mode == v1alpha1.MarketIdentityModeLocalRepo && strings.TrimSpace(status.Did) != "" && status.ObservedGeneration >= generation &&
		status.Phase != "" && status.Phase != v1alpha1.MarketIdentityPhasePending {
		if status.Phase == v1alpha1.MarketIdentityPhaseReady || status.Phase == v1alpha1.MarketIdentityPhaseAwaitingApproval {
			settleAssociation(&status, generation, id.Spec.Association, "the PDS reports "+status.Did)
		}
		return Result{Status: status}, nil
	}
	switch id.Spec.Mode {
	case v1alpha1.MarketIdentityModeLocalRepo:
		return a.localRepo(ctx, workspace, id, generation, status)
	case v1alpha1.MarketIdentityModePdsAccount:
		return a.pdsAccount(ctx, workspace, id, generation, status)
	case v1alpha1.MarketIdentityModeOAuthBrowser:
		return a.oauthBrowser(ctx, workspace, id, generation, status)
	case v1alpha1.MarketIdentityModeOAuthQr:
		return a.oauthQr(ctx, workspace, id, generation, status)
	case v1alpha1.MarketIdentityModeOAuthDelegated:
		return a.oauthDelegated(ctx, workspace, id, generation, status)
	case v1alpha1.MarketIdentityModeOAuthSessionProxy:
		return a.sessionProxy(ctx, workspace, id, generation, status)
	case "":
		return pending(status, generation, "UnknownMode", "the MarketIdentity has no mode; no DID is recorded"), nil
	default:
		return pending(status, generation, "UnknownMode", fmt.Sprintf("unknown mode %q; no DID is recorded", id.Spec.Mode)), nil
	}
}

func (a *Accounts) Forget(workspace, name string) {
	a.qr.Delete(qrKey(workspace, name))
	a.browser.Delete(qrKey(workspace, name))
	a.held.forget(workspace, name)
}

func (a *Accounts) localRepo(ctx context.Context, workspace string, id *v1alpha1.MarketIdentity, generation int64, status v1alpha1.MarketIdentityStatus) (Result, error) {
	endpoint := ""
	if a.endpoints != nil {
		if url, ok := a.endpoints(workspace, ComponentKind, ComponentName); ok {
			endpoint = strings.TrimSpace(url)
		}
	}
	if endpoint == "" {
		return pending(status, generation, "PdsNotReady", "waiting for the Pds/pds component endpoint in workspace "+workspace), nil
	}
	did, err := a.resolveDID(ctx, workspace, endpoint)
	if err != nil {
		return pending(status, generation, "PdsNotReady", "waiting for the PDS at "+endpoint+": "+err.Error()),
			fmt.Errorf("marketidentity: the PDS at %s: %w", endpoint, err)
	}
	composed := strings.TrimSpace(id.Spec.Did)
	if composed == "" {
		status.Did = ""
		return pending(status, generation, "AccountPending", "the PDS at "+endpoint+" ("+did+") is ready; waiting for the root to compose the account DID"), nil
	}
	status.Phase = v1alpha1.MarketIdentityPhaseReady
	status.ObservedGeneration = generation
	status.Did = composed
	if handle := strings.TrimSpace(id.Spec.Handle); handle != "" {
		status.Handle = handle
	}
	if pdsURL := strings.TrimSpace(id.Spec.PdsURL); pdsURL != "" {
		status.PdsURL = pdsURL
	}
	setAssociated(&status, generation, id.Spec.Association)
	settleAssociation(&status, generation, id.Spec.Association, "the PDS at "+endpoint+" reports "+did)
	return Result{Status: status}, nil
}

func settleAssociation(status *v1alpha1.MarketIdentityStatus, generation int64, mode v1alpha1.AssociationMode, ready string) {
	if mode != v1alpha1.AssociationSkip && !associatedTrue(status) {
		status.Phase = v1alpha1.MarketIdentityPhaseAwaitingApproval
		status.Message = "the PDS is ready; waiting for the operator's verified association claim"
		setReady(status, generation, false, "AwaitingAssociation", status.Message)
		return
	}
	status.Phase, status.Message = v1alpha1.MarketIdentityPhaseReady, ""
	setReady(status, generation, true, "Ready", ready)
}

func (a *Accounts) resolveDID(ctx context.Context, workspace, endpoint string) (string, error) {
	pds, err := a.localPDS(workspace, endpoint)
	if err != nil {
		return "", err
	}
	info, err := pds.Describe(ctx)
	if err != nil {
		return "", err
	}
	did := strings.TrimSpace(info.DID.String())
	if did == "" {
		return "", fmt.Errorf("the PDS reported no did")
	}
	return did, nil
}

func (a *Accounts) localPDS(workspace, endpoint string) (pdsclient.PDS, error) {
	endpoint = strings.TrimSpace(endpoint)
	if !strings.Contains(endpoint, "://") {
		endpoint = "http://" + endpoint
	}
	if a.resolver.Handlers != nil {
		return a.resolver.ResolveLocal(pdsclient.LocalRef{Workspace: workspace, Kind: ComponentKind, Name: ComponentName, Endpoint: endpoint})
	}
	return pdsclient.NewRemote(pdsclient.RemoteConfig{BaseURL: endpoint, HTTPClient: a.resolver.HTTPClient, TLS: a.resolver.TLS,
		Tokens: a.resolver.Tokens, DPoP: a.resolver.DPoP, Timeout: opt.Positive(a.resolver.Timeout, requestTimeout)})
}

func (a *Accounts) vaultFor(workspace string) SecretStore {
	if a.vault == nil {
		return nil
	}
	return a.vault(workspace)
}

func pending(status v1alpha1.MarketIdentityStatus, generation int64, reason, message string) Result {
	status.Phase = v1alpha1.MarketIdentityPhasePending
	status.ObservedGeneration = generation
	status.Message = message
	setReady(&status, generation, false, reason, message)
	return Result{Status: status}
}

func failed(status v1alpha1.MarketIdentityStatus, generation int64, reason, message string) Result {
	status.Phase = v1alpha1.MarketIdentityPhaseFailed
	status.ObservedGeneration = generation
	status.Message = message
	setReady(&status, generation, false, reason, message)
	return Result{Status: status}
}

func setReady(status *v1alpha1.MarketIdentityStatus, generation int64, ready bool, reason, message string) {
	conds.Set(&status.Conditions, generation, ConditionReady, ready, reason, message)
}

func refreshAfter(token string) time.Duration {
	parts := strings.Split(token, ".")
	if len(parts) < 2 {
		return 0
	}
	raw, err := base64.RawURLEncoding.DecodeString(parts[1])
	if err != nil {
		return 0
	}
	var claims struct {
		ExpiresAt int64 `json:"exp"`
	}
	if json.Unmarshal(raw, &claims) != nil || claims.ExpiresAt == 0 {
		return 0
	}
	return max(time.Until(time.Unix(claims.ExpiresAt, 0).Add(-oauthclient.AccessSkew)), minRefreshAfter)
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}