johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bidder

import (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"time"

	"github.com/gorilla/websocket"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/ingress"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	atprotooauth "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauth"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/component"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/hostport"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/httpjson"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/networkdial"
	core "github.com/johnandersen777/socialweb-computer-kcp/internal/market/bidder"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant"
)

const (
	SubmitRFPPath = "/xrpc/" + market.SubmitRFPNSID

	SubmitBidPath = "/xrpc/" + market.SubmitBidNSID

	SubmitAcceptPath = "/xrpc/" + market.SubmitAcceptNSID

	SubmitEventPath = "/xrpc/" + market.SubmitEventNSID

	relayTokenLifetime = 5 * time.Minute
)

type submittedRecord struct {
	URI string `json:"uri"`

	CID string `json:"cid"`

	Record map[string]any `json:"record"`

	Ingress []market.StrongRef `json:"ingress,omitempty"`
}

func (s *Service) routes() *http.ServeMux {
	mux := http.NewServeMux()
	if s.inboundXRPC() {
		mux.HandleFunc("POST "+SubmitRFPPath, s.handleSubmitRFP)
		mux.HandleFunc("POST "+SubmitBidPath, s.handleSubmitBid)
		mux.HandleFunc("POST "+SubmitAcceptPath, s.handleSubmitAccept)
		mux.HandleFunc("POST "+SubmitEventPath, s.handleSubmitEvent)
	}
	mux.HandleFunc("GET "+HealthPath, func(w http.ResponseWriter, r *http.Request) {
		httpjson.Write(w, http.StatusOK, map[string]any{"ok": true})
	})
	mux.HandleFunc("GET "+OAuthMetadataPath, s.oauthClientMetadata)
	mux.HandleFunc("GET /oauth/callback/{identity}", s.oauthCallback)
	return mux
}

func (s *Service) admit(w http.ResponseWriter, r *http.Request, lxm, recordURI string) (string, bool) {
	authorizer := s.authorizer.Load()
	if authorizer == nil {
		participant.WriteXRPCError(w, http.StatusServiceUnavailable, "Unavailable", "the bidder is not ready to authorize requests")
		return "", false
	}
	issuer, ok, status, err := authorizer.AuthorizeRequest(r, lxm, recordURI)
	if !ok {
		s.log.Warn("the bidder refused a market request", "lxm", lxm, "record", recordURI, "status", status, "serviceAuthHost", s.serviceAuthHost, "err", err.Error())
		participant.WriteXRPCError(w, status, participant.AuthCode(status), "invalid service-auth token")
	}
	return issuer, ok
}

func (s *Service) suspended(w http.ResponseWriter) bool {
	if !s.spec.SuspendParticipation {
		return false
	}
	participant.WriteXRPCError(w, http.StatusServiceUnavailable, "Suspended", "the bidder's participation is suspended")
	return true
}

func (s *Service) ready(w http.ResponseWriter, what string) *core.Bidder {
	bidder := s.bidder.Load()
	if bidder == nil {
		participant.WriteXRPCError(w, http.StatusServiceUnavailable, "Unavailable", "the bidder is not ready to "+what)
	}
	return bidder
}

func (s *Service) handleSubmitRFP(w http.ResponseWriter, r *http.Request) {
	var body struct {
		RFPURI string `json:"rfpUri"`
		RFPCID string `json:"rfpCid"`

		Ingress []market.StrongRef `json:"ingress"`
	}
	if err := participant.ReadSubmitBody(r, &body); err != nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
		return
	}
	if body.RFPURI == "" || body.RFPCID == "" {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", "missing rfpUri or rfpCid")
		return
	}
	issuer, ok := s.admit(w, r, market.SubmitRFPNSID, body.RFPURI)
	if !ok || s.suspended(w) {
		return
	}
	bidder := s.ready(w, "bid")
	if bidder == nil {
		return
	}
	name, err := bidder.SubmitRFP(r.Context(), body.RFPURI, body.RFPCID, issuer)
	if errors.Is(err, core.ErrRFPClosed) {
		participant.WriteXRPCError(w, http.StatusConflict, "RFPClosed", err.Error())
		return
	}
	if err != nil {
		s.log.Warn("recording the bid intent failed", "rfp", body.RFPURI, "err", err)
		participant.WriteXRPCError(w, http.StatusInternalServerError, "InternalError", err.Error())
		return
	}
	out := map[string]any{"ok": true, "name": name, "phase": string(v1alpha1.BidPhaseDeciding)}
	if len(body.Ingress) > 0 {
		if refs, err := bidder.AnswerIngressRecords(r.Context(), body.Ingress, issuer); err != nil {
			s.log.Warn("bidder_webrtc_answer_failed", "rfp", body.RFPURI, "requester", issuer, "err", err)
		} else if len(refs) > 0 {
			list := make([]map[string]any, 0, len(refs))
			for _, ref := range refs {
				list = append(list, ref.Map())
			}
			out["ingress"] = list
		}
	}
	httpjson.Write(w, http.StatusOK, out)
}

func (s *Service) handleSubmitBid(w http.ResponseWriter, r *http.Request) {
	var body submittedRecord
	if err := participant.ReadSubmitBody(r, &body); err != nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
		return
	}
	if body.URI == "" || body.CID == "" || body.Record == nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", "missing uri, cid, or record")
		return
	}
	if _, ok := s.admit(w, r, market.SubmitBidNSID, body.URI); !ok {
		return
	}
	if _, err := market.ParseBid(body.Record); err != nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
		return
	}
	httpjson.Write(w, http.StatusOK, map[string]any{"ok": true})
}

func (s *Service) handleSubmitAccept(w http.ResponseWriter, r *http.Request) {
	var body struct {
		AcceptURI string `json:"acceptUri"`
		AcceptCID string `json:"acceptCid"`

		Ingress []market.StrongRef `json:"ingress,omitempty"`
	}
	if err := participant.ReadSubmitBody(r, &body); err != nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
		return
	}
	if body.AcceptURI == "" || body.AcceptCID == "" {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", "missing acceptUri or acceptCid")
		return
	}
	issuer, ok := s.admit(w, r, market.SubmitAcceptNSID, body.AcceptURI)
	if !ok {
		return
	}
	bidder := s.ready(w, "accept")
	if bidder == nil {
		return
	}
	id, err := bidder.SubmitAccept(r.Context(), body.AcceptURI, body.AcceptCID, issuer)
	if err != nil {
		participant.WriteXRPCError(w, http.StatusInternalServerError, "InternalError", "the provisioning intent could not be recorded: "+err.Error())
		return
	}
	httpjson.Write(w, http.StatusAccepted, map[string]any{"ok": true, "id": id, "acceptUri": body.AcceptURI, "acceptCid": body.AcceptCID})
}

func (s *Service) handleSubmitEvent(w http.ResponseWriter, r *http.Request) {
	var body submittedRecord
	if err := participant.ReadSubmitBody(r, &body); err != nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
		return
	}
	if body.URI == "" || body.CID == "" || body.Record == nil {
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", "missing uri, cid, or record")
		return
	}
	issuer, ok := s.admit(w, r, market.SubmitEventNSID, body.URI)
	if !ok {
		return
	}
	bidder := s.ready(w, "accept events")
	if bidder == nil {
		return
	}
	err := bidder.SubmitEvent(r.Context(), body.URI, body.CID, body.Record, issuer)
	switch {
	case err == nil:
		httpjson.Write(w, http.StatusOK, map[string]any{"ok": true})
	case errors.Is(err, core.ErrEventForbidden):
		participant.WriteXRPCError(w, http.StatusForbidden, "Forbidden", "not the accept author")
	case errors.Is(err, core.ErrEventInvalid):
		participant.WriteXRPCError(w, http.StatusBadRequest, "InvalidRequest", err.Error())
	default:
		s.log.Warn("recording the event intent failed", "event", body.URI, "err", err)
		participant.WriteXRPCError(w, http.StatusServiceUnavailable, "Unavailable", err.Error())
	}
}

func (s *Service) oauthClientMetadata(w http.ResponseWriter, r *http.Request) {
	base := "https://" + r.Host
	httpjson.Write(w, http.StatusOK, map[string]any{
		"client_id":                  base + "/oauth-client-metadata.json",
		"client_name":                "socialweb-computer bidder",
		"redirect_uris":              []string{base + "/oauth/callback/bidder"},
		"grant_types":                []string{"authorization_code", "refresh_token"},
		"response_types":             []string{"code"},
		"scope":                      atprotooauth.ScopeString(atprotooauth.BidderOAuthScope),
		"token_endpoint_auth_method": "none",
		"application_type":           "web",
		"dpop_bound_access_tokens":   true,
	})
}

func (s *Service) oauthCallback(w http.ResponseWriter, r *http.Request) {
	name := r.PathValue("identity")
	s.log.Info("oauth callback received", "identity", name, "state", r.URL.Query().Get("state"))
	handler, ok := s.deps.OAuthCallbacks.Lookup(name)
	s.log.Info("oauth callback lookup", "identity", name, "found", ok)
	if !ok {
		http.Error(w, "unknown identity", http.StatusNotFound)
		return
	}
	if err := handler(r.Context(), w, r); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write([]byte("signed in"))
}

func (s *Service) startRelay(ctx context.Context) error {
	ingressHost := s.network().IngressProxyHost
	if ingressHost == "" {
		return nil
	}
	if s.stateDir() == "" {
		return errors.New("the relay names no keypair and there is no state dir")
	}
	signer, err := market.LoadOrCreateSignerFile(s.stateDir() + "/market-relay.jwk")
	if err != nil {
		return err
	}
	did, err := signer.PublicDID()
	if err != nil {
		return err
	}
	aud := "did:web:" + hostport.URLHostname(ingressHost)
	config := ingress.SubscriberConfig{
		Label:   "market",
		Keypair: relayKeypair{signer: signer, did: did},
		GetServiceAuthToken: func(ctx context.Context, nsid string) (string, error) {
			return ingress.SignServiceAuth(signer.PrivateKey, did, aud, nsid, relayTokenLifetime)
		},
		IngressProxyHost: ingressHost,
		HandleRequest:    s.forwardRelay,
	}
	if dialAddress := ingressConnectHost(s.network()); dialAddress != "" {
		hosts, err := networkdial.NewHostMap([]networkdial.Host{{Name: hostport.URLHostname(ingressHost), Address: dialAddress}})
		if err != nil {
			return err
		}
		transport, err := networkdial.NewHTTPTransportWithOptions(networkdial.DialOptions{Hosts: hosts})
		if err != nil {
			return err
		}
		config.HTTPClient = &http.Client{Transport: transport, Timeout: DefaultCallTimeout}
		var dialer *websocket.Dialer
		if dialer, err = networkdial.NewWebSocketDialerWithOptions(networkdial.DialOptions{Hosts: hosts}); err != nil {
			return err
		}
		config.Dialer = dialer
	}
	subscriber, err := ingress.NewSubscriber(ctx, config)
	if err != nil {
		return err
	}
	s.mu.Lock()
	s.relay = subscriber
	s.mu.Unlock()
	return nil
}

func (s *Service) relayEndpoint() (component.Endpoint, bool) {
	s.mu.Lock()
	relay := s.relay
	s.mu.Unlock()
	host := s.network().IngressProxyHost
	if relay == nil || host == "" {
		return component.Endpoint{}, false
	}
	scheme := "https"
	if base := strings.TrimSpace(s.spec.Network.GuestTLSURL()); base != "" {
		parsed, err := url.Parse(base)
		if err != nil || parsed.Host == "" {
			return component.Endpoint{}, false
		}
		scheme, host = parsed.Scheme, parsed.Host
	} else if strings.HasPrefix(host, "127.0.0.1") || strings.HasPrefix(host, "localhost") {
		scheme = "http"
	}
	protocol := component.ProtocolHTTPS
	if scheme == "http" {
		protocol = component.ProtocolHTTP
	}
	return component.Endpoint{Name: "relay", Protocol: protocol, URL: scheme + "://" + relay.Subdomain() + "." + host}, true
}

func (s *Service) forwardRelay(ctx context.Context, request *ingress.Request) (*ingress.Response, error) {
	if request == nil {
		return nil, errors.New("the relay request is empty")
	}
	path := cmp.Or(request.Path, "/")
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
	}
	if len(request.Params) > 0 {
		values := url.Values{}
		for key, value := range request.Params {
			values.Set(key, value)
		}
		path += "?" + values.Encode()
	}
	body := request.RawBody
	if body == "" && len(request.Body) > 0 {
		body = string(request.Body)
	}
	req, err := http.NewRequestWithContext(ctx, request.Method, "http://relay.local"+path, strings.NewReader(body))
	if err != nil {
		return nil, err
	}
	for key, value := range request.Headers {
		req.Header.Set(key, value)
	}
	recorder := httptest.NewRecorder()
	s.mux.ServeHTTP(recorder, req)
	result := recorder.Result()
	defer result.Body.Close()
	raw, err := io.ReadAll(io.LimitReader(result.Body, RecordReplyLimit))
	if err != nil {
		return nil, err
	}
	return &ingress.Response{Status: result.StatusCode, Body: json.RawMessage(raw), ContentType: result.Header.Get("Content-Type")}, nil
}

type relayKeypair struct {
	signer *market.Signer

	did string
}

func (k relayKeypair) DID() string { return k.did }

func (k relayKeypair) Sign(data []byte) ([]byte, error) {
	return market.SignBytes(data, k.signer.PrivateKey)
}