johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package requester

import (
	"context"
	"errors"
	"slices"
	"strconv"
	"strings"
	"sync"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
)

const (
	sourceConfigured = "configured"

	sourceRelayPrefix = "relay:"

	sourceVouched = "vouched"

	sourceOperated = "operated"

	bidderAssociateService = "bidder_associate"

	requesterAssociateService = "requester_associate"

	trustListLimit = 100

	maxTrustRepos = 64
)

type candidateSource struct {
	name string

	dids []string

	named bool
}

func didValue(value any) string {
	s, _ := value.(string)
	if s = strings.TrimSpace(s); strings.HasPrefix(s, "did:") {
		return s
	}
	return ""
}

func associationRecord(value map[string]any) (service, challenge, keyID string) {
	service, _ = value["service"].(string)
	return strings.TrimSpace(service), didValue(value["challenge"]), didValue(value["keyId"])
}

func (m *HTTPMarket) listRepo(ctx context.Context, did, collection string) []market.ListedRecord {
	records, err := listAll(ctx, m.Reader, did, collection, trustListLimit)
	if err != nil && !errors.Is(err, market.ErrNotFound) && ctx.Err() == nil {
		m.log().Debug("requester_discovery_list_failed", "did", did, "collection", collection, "err", err)
	}
	return records
}

func (m *HTTPMarket) selfDIDs() []string {
	return BidderDIDList([]string{m.RepoDID})
}

func (m *HTTPMarket) operatorsOf(ctx context.Context, self string) []string {
	var out []string
	for _, record := range m.listRepo(ctx, self, market.BadgeBlueKeysNSID) {
		service, challenge, keyID := associationRecord(record.Value)
		if service != bidderAssociateService && service != requesterAssociateService {
			continue
		}
		switch {
		case keyID == self && challenge != "" && challenge != self:
			out = append(out, challenge)
		case challenge == self && keyID != "" && keyID != self:
			out = append(out, keyID)
		}
	}
	return out
}

func (m *HTTPMarket) vouchedBy(ctx context.Context, did string) []string {
	var out []string
	for _, record := range m.listRepo(ctx, did, market.VouchNSID) {
		if kind, _ := record.Value["kind"].(string); kind == "denounce" {
			continue
		}
		if rkey := record.URI[strings.LastIndex(record.URI, "/")+1:]; strings.HasPrefix(rkey, "did:") && rkey != did {
			out = append(out, rkey)
		}
	}
	return out
}

func associates(value map[string]any, operator string) string {
	service, challenge, keyID := associationRecord(value)
	if service != bidderAssociateService || challenge != operator || keyID == operator {
		return ""
	}
	return keyID
}

func (m *HTTPMarket) operatedBy(ctx context.Context, operator string) []string {
	var candidates []string
	for _, record := range m.listRepo(ctx, operator, market.BadgeBlueKeysNSID) {
		if did := associates(record.Value, operator); did != "" {
			candidates = append(candidates, did)
		}
	}
	var out []string
	for _, did := range BidderDIDList(candidates) {
		mirrored := slices.ContainsFunc(m.listRepo(ctx, did, market.BadgeBlueKeysNSID), func(record market.ListedRecord) bool {
			return associates(record.Value, operator) == did
		})
		if !mirrored {
			m.log().Info("requester_operated_unmirrored", "operator", operator, "bidder", did)
			continue
		}
		out = append(out, did)
	}
	return out
}

func eachParallel(items []string, fn func(string) []string) []string {
	var mu sync.Mutex
	var wg sync.WaitGroup
	var out []string
	for _, item := range items {
		wg.Go(func() {
			found := fn(item)
			mu.Lock()
			out = append(out, found...)
			mu.Unlock()
		})
	}
	wg.Wait()
	return BidderDIDList(out)
}

func capped(dids []string) []string {
	if len(dids) > maxTrustRepos {
		return dids[:maxTrustRepos]
	}
	return dids
}

func (m *HTTPMarket) relaySources(ctx context.Context, roots []string) ([]candidateSource, int) {
	relays := append([]string(nil), m.RelayURLs...)
	auto := 0
	for _, did := range roots {
		for _, relay := range market.RelayURLsOf(ctx, m.Reader, did) {
			relay = strings.TrimRight(relay, "/")
			if !slices.Contains(relays, relay) {
				relays = append(relays, relay)
				auto++
			}
		}
	}
	out := make([]candidateSource, len(relays))
	var wg sync.WaitGroup
	for i, relay := range relays {
		wg.Go(func() {
			out[i] = candidateSource{name: sourceRelayPrefix + relay,
				dids: market.DiscoverBiddersFromRelay(ctx, relay, market.OfferingNSID, bidderDiscoveryTimeout, m.client(), market.NoopLog)}
		})
	}
	wg.Wait()
	return out, auto
}

func (m *HTTPMarket) candidateSources(ctx context.Context) ([]candidateSource, int) {
	self := m.selfDIDs()
	operators := BidderDIDList(append(append([]string{m.OperatorDID}, self...), eachParallel(self, func(did string) []string {
		return m.operatorsOf(ctx, did)
	})...))
	roots := BidderDIDList(append(append([]string(nil), self...), m.OperatorDID))
	var relays []candidateSource
	var auto int
	var vouched []string
	var wg sync.WaitGroup
	wg.Go(func() { relays, auto = m.relaySources(ctx, roots) })
	wg.Go(func() {
		vouched = capped(eachParallel(operators, func(did string) []string { return m.vouchedBy(ctx, did) }))
	})
	wg.Wait()
	operated := eachParallel(capped(BidderDIDList(append(append([]string(nil), operators...), vouched...))), func(did string) []string {
		return m.operatedBy(ctx, did)
	})
	out := []candidateSource{{name: sourceConfigured, dids: BidderDIDList(m.BidderDIDs), named: true}}
	out = append(out, relays...)
	out = append(out, candidateSource{name: sourceVouched, dids: vouched}, candidateSource{name: sourceOperated, dids: operated})
	return out, auto
}

func sourceCounts(sources []candidateSource, auto int) string {
	parts := make([]string, 0, len(sources)+1)
	for _, source := range sources {
		parts = append(parts, source.name+"="+strconv.Itoa(len(source.dids)))
	}
	parts = append(parts, "relaysFromRecords="+strconv.Itoa(auto))
	return strings.Join(parts, " ")
}