johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bidder

import (
	"context"
	"errors"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

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

func (b *Bidder) Run(ctx context.Context) {
	defer b.wg.Wait()
	b.wg.Go(func() { b.bindBadgeBlueKey(ctx) })
	if b.cfg.Suspend {
		b.withdraw(ctx)
		return
	}
	b.serveMarket(ctx)
}

func (b *Bidder) bindBadgeBlueKey(ctx context.Context) {
	b.badgeMu.Lock()
	defer b.badgeMu.Unlock()
	if b.badgeBound.Load() {
		return
	}
	key, err := b.signer.PublicDID()
	if err != nil {
		b.log.Warn("reading the attestation key DID failed", "err", err)
		return
	}
	if _, err := market.CreateBadgeBlueKeysRecord(ctx, b.records, market.BadgeBlueKeysRecordOptions{
		DID:     b.cfg.DID,
		KeyID:   key,
		Service: "bidder_attestation",
	}); err != nil {
		if ctx.Err() == nil {
			b.log.Warn("binding the badge.blue key failed", "err", err)
		}
		return
	}
	b.badgeBound.Store(true)
	b.log.Info("the bidder's badge.blue attestation key is bound", "did", b.cfg.DID, "key", key)
}

func (b *Bidder) serveMarket(ctx context.Context) {
	if err := b.ensureOffering(ctx); err != nil {
		b.log.Warn("creating the market offering failed", "err", err)
	}
	if err := b.beat(ctx); err != nil {
		b.log.Warn("sending the market heartbeat failed", "err", err)
	}
	b.verifyVisibility(ctx)
	if b.cfg.Heartbeat <= 0 {
		b.log.Warn("the market heartbeat is disabled", "period", b.cfg.Heartbeat)
		return
	}
	heartbeat := time.NewTicker(b.cfg.Heartbeat)
	defer heartbeat.Stop()
	refreshed := b.now()
	for {
		select {
		case <-ctx.Done():
			return
		case <-heartbeat.C:
		}
		if !b.badgeBound.Load() {
			b.bindBadgeBlueKey(ctx)
		}
		if b.currentOffering() == "" {
			if err := b.ensureOffering(ctx); err != nil && ctx.Err() == nil {
				b.log.Warn("creating the market offering failed", "err", err)
			}
		}
		if err := b.beat(ctx); err != nil && ctx.Err() == nil {
			b.log.Warn("sending the market heartbeat failed", "err", err)
		}
		if b.cfg.OfferingRefresh <= 0 || b.now().Sub(refreshed) < b.cfg.OfferingRefresh {
			continue
		}
		refreshed = b.now()
		if err := b.ensureOffering(ctx); err != nil {
			if ctx.Err() == nil {
				b.log.Warn("refreshing the market offering failed", "err", err)
			}
			continue
		}
		b.verifyVisibility(ctx)
	}
}

func (b *Bidder) withdraw(ctx context.Context) {
	b.offerWrite.Lock()
	defer b.offerWrite.Unlock()
	prior, readErr := b.selfOffering(ctx)
	if readErr != nil {
		b.log.Warn("reading the market offering to withdraw failed; it is withdrawn anyway", "err", readErr)
	}
	if prior != nil && market.OfferingWithdrawn(*prior) {
		return
	}
	createdAt := ""
	if prior != nil {
		createdAt = prior.CreatedAt
	}
	record, err := market.BuildWithdrawnOffering(createdAt, b.now().UTC().Format(time.RFC3339))
	if err != nil {
		b.log.Warn("building the withdrawn market offering failed", "err", err)
		return
	}
	written := false
	if prior != nil || readErr != nil {
		written = b.withdrawSelf(ctx, record)
	}
	if err := b.findLegacyOffering(ctx); err != nil {
		b.log.Warn("reading the heartbeat to withdraw the legacy market offering failed", "err", err)
		if !written && prior == nil {
			b.withdrawSelf(ctx, record)
		}
		return
	}
	if b.legacyOffering == "" {
		return
	}
	if !written {
		b.withdrawSelf(ctx, record)
	}
	b.mirrorLegacy(ctx, record)
}

func (b *Bidder) withdrawSelf(ctx context.Context, record market.Offering) bool {
	ref, err := b.putRecord(ctx, market.OfferingRKey, record)
	if err != nil {
		b.log.Warn("withdrawing the market offering failed", "err", err)
		return false
	}
	b.log.Info("the bidder offering is withdrawn while its participation is suspended", "offering", ref.URI)
	return true
}

func (b *Bidder) setOffering(uri string) {
	b.offerMu.Lock()
	b.offeringURI = uri
	b.offerMu.Unlock()
}

func (b *Bidder) ensureOffering(ctx context.Context) error {
	b.offerWrite.Lock()
	defer b.offerWrite.Unlock()
	return b.ensureOfferingLocked(ctx, false)
}

func (b *Bidder) ensureOfferingLocked(ctx context.Context, touch bool) error {
	appliesTo := []string{market.ComputeVMNSID}
	now := b.now().UTC().Format(time.RFC3339)
	prior, err := b.selfOffering(ctx)
	if err != nil {
		return err
	}
	if err := b.findLegacyOffering(ctx); err != nil {
		return err
	}
	createdAt := now
	if prior != nil {
		createdAt = prior.CreatedAt
		if !touch && prior.WithdrawnAt == "" && prior.EndpointURL == b.cfg.EndpointURL && sameStrings(prior.AppliesTo, appliesTo) {
			if b.legacyStale {
				b.mirrorLegacy(ctx, *prior)
			}
			b.setOffering(b.heartbeatOffering())
			return nil
		}
	}
	record, err := market.BuildOffering(b.cfg.EndpointURL, appliesTo, nil, createdAt, now)
	if err != nil {
		return err
	}
	uri, err := b.putOffering(ctx, record)
	if err != nil {
		return err
	}
	b.setOffering(b.heartbeatOffering())
	switch {
	case touch:
		b.log.Info("the bidder offering is refreshed for the relay", "offering", uri, "legacy", b.legacyOffering)
	case prior == nil:
		b.log.Info("the bidder offering is written", "offering", uri, "endpoint", b.cfg.EndpointURL, "legacy", b.legacyOffering)
	default:
		b.log.Info("the bidder offering is corrected", "offering", uri, "endpoint", b.cfg.EndpointURL, "legacy", b.legacyOffering)
	}
	return nil
}

func (b *Bidder) TouchOffering(ctx context.Context) {
	if b.cfg.Suspend {
		return
	}
	if err := b.touchOffering(ctx); err != nil {
		b.log.Warn("republishing the offering after the relay crawl failed", "err", err.Error())
	}
}

func (b *Bidder) touchOffering(ctx context.Context) error {
	b.offerWrite.Lock()
	defer b.offerWrite.Unlock()
	return b.ensureOfferingLocked(ctx, true)
}

func (b *Bidder) selfOffering(ctx context.Context) (*market.Offering, error) {
	rec, err := b.records.GetRecord(ctx, b.cfg.DID, market.OfferingNSID, market.OfferingRKey)
	if errors.Is(err, market.ErrNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	offering, err := market.ParseOffering(rec.Value)
	if err != nil {
		b.log.Warn("the bidder's own market offering is invalid; it is written again", "offering", rec.URI, "err", err)
		return nil, nil
	}
	return offering, nil
}

func (b *Bidder) findLegacyOffering(ctx context.Context) error {
	if b.legacyChecked {
		return nil
	}
	rkey, err := market.HeartbeatOfferingRKey(ctx, b.records, b.cfg.DID)
	switch {
	case errors.Is(err, market.ErrNotFound), errors.Is(err, market.ErrInvalidRecord):
		b.legacyChecked = true
		return nil
	case err != nil:
		return err
	}
	if _, err := syntax.ParseTID(rkey); rkey == market.OfferingRKey || err != nil {
		b.legacyChecked = true
		return nil
	}
	rec, err := b.records.GetRecord(ctx, b.cfg.DID, market.OfferingNSID, rkey)
	switch {
	case errors.Is(err, market.ErrNotFound):
	case err != nil:
		return err
	case rec.CID == "":
		b.log.Warn("the legacy market offering has no cid; it is not mirrored", "legacy", rkey)
	default:
		b.legacyOffering = rkey
		b.legacyCID = rec.CID
		b.log.Info("the bidder mirrors its offering to the legacy record its heartbeat named", "legacy", rkey)
	}
	b.legacyChecked = true
	return nil
}

func (b *Bidder) heartbeatOffering() string {
	if b.legacyOffering != "" {
		return "at://" + b.cfg.DID + "/" + market.OfferingNSID + "/" + b.legacyOffering
	}
	return market.OfferingURI(b.cfg.DID)
}

func (b *Bidder) putOffering(ctx context.Context, record market.Offering) (string, error) {
	ref, err := b.putRecord(ctx, market.OfferingRKey, record)
	if err != nil {
		return "", err
	}
	b.mirrorLegacy(ctx, record)
	return ref.URI, nil
}

func (b *Bidder) mirrorLegacy(ctx context.Context, record market.Offering) {
	if b.legacyOffering == "" {
		return
	}
	value, err := market.EncodeRecord(record)
	if err != nil {
		b.log.Warn("encoding the legacy market offering failed", "legacy", b.legacyOffering, "err", err)
		b.legacyStale = true
		return
	}
	ref, err := b.records.PutRecordSwap(ctx, b.cfg.DID, market.OfferingNSID, b.legacyOffering, value, b.legacyCID)
	switch {
	case errors.Is(err, market.ErrSwapMismatch), errors.Is(err, market.ErrNotFound):
		b.log.Warn("the legacy market offering changed or is gone; the mirror stops and the heartbeat names self", "legacy", b.legacyOffering, "err", err)
		b.legacyOffering, b.legacyCID, b.legacyStale = "", "", false
	case err != nil:
		if ctx.Err() == nil {
			b.log.Warn("updating the legacy market offering failed", "legacy", b.legacyOffering, "err", err)
		}
		b.legacyStale = true
	default:
		b.legacyCID, b.legacyStale = ref.CID, false
	}
}

func (b *Bidder) currentOffering() string {
	b.offerMu.Lock()
	defer b.offerMu.Unlock()
	return b.offeringURI
}

func (b *Bidder) beat(ctx context.Context) error {
	b.offerMu.Lock()
	offering := b.offeringURI
	b.offerMu.Unlock()
	if offering == "" {
		return nil
	}
	record, err := market.BuildHeartbeat(offering, b.now().UTC().Format(time.RFC3339))
	if err != nil {
		return err
	}
	_, err = b.putRecord(ctx, market.HeartbeatRKey, record)
	return err
}

func (b *Bidder) putRecord(ctx context.Context, rkey string, record market.Record) (market.StrongRef, error) {
	value, err := market.EncodeRecord(record)
	if err != nil {
		return market.StrongRef{}, err
	}
	return b.records.PutRecord(ctx, b.cfg.DID, record.NSID(), rkey, value)
}

func (b *Bidder) verifyVisibility(ctx context.Context) {
	if len(b.cfg.Relays) == 0 {
		return
	}
	relays := append([]string(nil), b.cfg.Relays...)
	b.wg.Go(func() {
		market.VerifyRelayVisibility(ctx, market.RelayVisibilityOptions{
			RelayURLs:  relays,
			BidderDID:  b.cfg.DID,
			Collection: market.OfferingNSID,
			Log:        func(level, msg string, fields map[string]any) { b.log.Info(msg, "fields", fields) },
		})
	})
}

func sameStrings(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if strings.TrimSpace(a[i]) != strings.TrimSpace(b[i]) {
			return false
		}
	}
	return true
}