1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
package request
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"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"
"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
"github.com/johnandersen777/socialweb-computer-kcp/internal/webrtcproxy"
)
const contractEventPrefix = "cli-"
var _ webrtcproxy.Signaling = (*ContractSignaling)(nil)
type ContractSignaling struct {
k *Kcp
contract string
session string
}
func (k *Kcp) Signaling() *ContractSignaling {
var b [6]byte
_, _ = rand.Read(b[:])
return &ContractSignaling{k: k, contract: k.ContractName(), session: hex.EncodeToString(b[:])}
}
func (s *ContractSignaling) eventName(suffix string) string {
return contractEventPrefix + s.session + "-" + suffix
}
func (s *ContractSignaling) create(ctx context.Context, suffix string, spec v1alpha1.ContractEventSpec) error {
spec.ContractRef = commonv1.LocalObjectRef{Name: s.contract}
spec.Direction = v1alpha1.ContractEventOutbound
event := v1alpha1.ContractEvent{TypeMeta: typeMeta("ContractEvent"),
ObjectMeta: metav1.ObjectMeta{Name: s.eventName(suffix)}, Spec: spec}
body, err := kube.ApplyBody(&event, false)
if err != nil {
return err
}
_, err = kube.Create[kube.Doc](ctx, s.k.dyn.Resource(v1alpha1.ContractEventGVR), body, metav1.CreateOptions{FieldManager: FieldManager})
return err
}
func (s *ContractSignaling) PublishKey(ctx context.Context, tag, key string) error {
return s.create(ctx, "key", v1alpha1.ContractEventSpec{NSID: v1alpha1.ContractEventNSIDGuestPublicKey,
Kind: market.GuestPublicKeyKindAgeX25519, Tag: tag, Key: key})
}
func (s *ContractSignaling) PublishOffer(ctx context.Context, tag, ciphertext string) error {
return s.create(ctx, "offer", v1alpha1.ContractEventSpec{NSID: v1alpha1.ContractEventNSIDGuestSignalWebRTC,
Kind: market.GuestSignalKindOffer, Tag: tag, Ciphertext: ciphertext})
}
func (s *ContractSignaling) PublishR2Start(ctx context.Context, start v1alpha1.ContractEventR2Start) (string, error) {
suffix := "r2-" + start.Name
return s.eventName(suffix), s.create(ctx, suffix, v1alpha1.ContractEventSpec{NSID: v1alpha1.ContractEventNSIDGuestR2Start, R2Start: &start})
}
func (s *ContractSignaling) PublishAgentStart(ctx context.Context) (string, error) {
return s.eventName("agent"), s.create(ctx, "agent", v1alpha1.ContractEventSpec{NSID: v1alpha1.ContractEventNSIDGuestAgentStart})
}
func (k *Kcp) WaitContractEvent(ctx context.Context, name string, timeout time.Duration, done func(*v1alpha1.ContractEvent) bool) (*v1alpha1.ContractEvent, error) {
return kube.Await(ctx, k.dyn.Resource(v1alpha1.ContractEventGVR), name, timeout, func(e *v1alpha1.ContractEvent) (bool, error) {
return e != nil && done(e), nil
})
}
func (s *ContractSignaling) inbound(e v1alpha1.ContractEvent, nsid, kind, tag string) bool {
return e.DeletionTimestamp == nil && e.Spec.Direction == v1alpha1.ContractEventInbound && e.Spec.ContractRef.Name == s.contract &&
e.Spec.NSID == nsid && e.Spec.Kind == kind && e.Spec.Tag == tag && e.Status.Phase == v1alpha1.ContractEventReceived
}
func (s *ContractSignaling) Guest(ctx context.Context, tag string) (webrtcproxy.Guest, error) {
var guest webrtcproxy.Guest
var failed error
_, err := kube.Await(ctx, s.k.dyn.Resource(v1alpha1.ComputeContractGVR), s.contract, 0, func(c *v1alpha1.ComputeContract) (bool, error) {
if c == nil {
return false, nil
}
switch c.Status.Phase {
case v1alpha1.ComputeContractReleased, v1alpha1.ComputeContractFailed:
failed = fmt.Errorf("the contract is %s: %s", c.Status.Phase, c.Status.Message)
return true, nil
}
if c.Status.Contract == nil {
return false, nil
}
for _, n := range c.Status.Contract.Networks {
if n.Tag == tag && n.Kind == v1alpha1.NetworkKindWebRTC && strings.TrimSpace(n.Address) != "" {
guest.Fingerprint = n.Address
return true, nil
}
}
return false, nil
})
if failed != nil {
return guest, failed
}
if err != nil {
return guest, fmt.Errorf("no onNetwork kind webrtc for the tag in status.contract.networks: %w", err)
}
_, err = kube.AwaitList(ctx, s.k.dyn.Resource(v1alpha1.ContractEventGVR), metav1.ListOptions{}, 0, func(items []v1alpha1.ContractEvent) (bool, error) {
var newest *v1alpha1.ContractEvent
for i := range items {
e := &items[i]
if !s.inbound(*e, v1alpha1.ContractEventNSIDGuestPublicKey, market.GuestPublicKeyKindAgeX25519, tag) || e.Spec.Key == "" {
continue
}
if newest == nil || newest.CreationTimestamp.Before(&e.CreationTimestamp) {
newest = e
}
}
if newest == nil {
return false, nil
}
guest.Key = newest.Spec.Key
return true, nil
})
if err != nil {
return guest, fmt.Errorf("no inbound guest publicKey ContractEvent for the tag: %w", err)
}
return guest, nil
}
func (s *ContractSignaling) Answers(ctx context.Context, tag string, each func(ciphertext string) bool) error {
seen := map[string]bool{}
_, err := kube.AwaitList(ctx, s.k.dyn.Resource(v1alpha1.ContractEventGVR), metav1.ListOptions{}, 0, func(items []v1alpha1.ContractEvent) (bool, error) {
for _, e := range items {
key := e.Name + "/" + string(e.UID)
if seen[key] || !s.inbound(e, v1alpha1.ContractEventNSIDGuestSignalWebRTC, market.GuestSignalKindAnswer, tag) || e.Spec.Ciphertext == "" {
continue
}
seen[key] = true
if each(e.Spec.Ciphertext) {
return true, nil
}
}
return false, nil
})
if errors.Is(err, kube.ErrAwaitTimeout) {
return context.DeadlineExceeded
}
return err
}
