alice.pds.demo.boltless.dev
This repository has no description
3.6 kB
152 lines
1package knotmirror
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "fmt"
8 "log/slog"
9 "net/netip"
10 "net/url"
11 "time"
12
13 "tangled.org/core/api/tangled"
14 "tangled.org/core/knotmirror/config"
15 "tangled.org/core/knotmirror/db"
16 "tangled.org/core/knotmirror/knotstream"
17 "tangled.org/core/knotmirror/models"
18 "tangled.org/core/log"
19 "tangled.org/core/tapc"
20)
21
22type Tap struct {
23 logger *slog.Logger
24 cfg *config.Config
25 tap tapc.Client
26 db *sql.DB
27 gitm GitMirrorManager
28 ks *knotstream.KnotStream
29}
30
31func NewTapClient(l *slog.Logger, cfg *config.Config, db *sql.DB, gitm GitMirrorManager, ks *knotstream.KnotStream) *Tap {
32 return &Tap{
33 logger: log.SubLogger(l, "tapclient"),
34 cfg: cfg,
35 tap: tapc.NewClient(cfg.TapUrl, ""),
36 db: db,
37 gitm: gitm,
38 ks: ks,
39 }
40}
41
42func (t *Tap) Start(ctx context.Context) {
43 // TODO: better reconnect logic
44 go func() {
45 for {
46 t.tap.Connect(ctx, &tapc.SimpleIndexer{
47 EventHandler: t.processEvent,
48 })
49 time.Sleep(time.Second)
50 }
51 }()
52}
53
54func (t *Tap) processEvent(ctx context.Context, evt tapc.Event) error {
55 l := t.logger.With("component", "tapIndexer")
56
57 var err error
58 switch evt.Type {
59 case tapc.EvtRecord:
60 switch evt.Record.Collection.String() {
61 case tangled.RepoNSID:
62 err = t.processRepo(ctx, evt.Record)
63 }
64 }
65
66 if err != nil {
67 l.Error("failed to process message. will retry later", "event.ID", evt.ID, "err", err)
68 return err
69 }
70 return nil
71}
72
73func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error {
74 switch evt.Action {
75 case tapc.RecordCreateAction, tapc.RecordUpdateAction:
76 record := tangled.Repo{}
77 if err := json.Unmarshal(evt.Record, &record); err != nil {
78 return fmt.Errorf("parsing record: %w", err)
79 }
80
81 status := models.RepoStatePending
82 errMsg := ""
83 u, err := url.Parse("http://" + record.Knot) // parsing with fake scheme
84 if err != nil {
85 status = models.RepoStateSuspended
86 errMsg = "failed to parse knot url"
87 } else if t.cfg.KnotSSRF && isPrivate(u.Hostname()) {
88 status = models.RepoStateSuspended
89 errMsg = "suspending non-public knot"
90 }
91
92 repo := &models.Repo{
93 Did: evt.Did,
94 Rkey: evt.Rkey,
95 Cid: evt.CID,
96 Name: record.Name,
97 KnotDomain: record.Knot,
98 State: status,
99 ErrorMsg: errMsg,
100 RetryAfter: 0, // clear retry info
101 RetryCount: 0,
102 }
103
104 if evt.Action == tapc.RecordUpdateAction {
105 exist, err := t.gitm.Exist(repo)
106 if err != nil {
107 return fmt.Errorf("checking git repo existance: %w", err)
108 }
109 if exist {
110 // update git repo remote url
111 if err := t.gitm.RemoteSetUrl(ctx, repo); err != nil {
112 return fmt.Errorf("updating git repo remote url: %w", err)
113 }
114 }
115 }
116
117 if err := db.UpsertRepo(ctx, t.db, repo); err != nil {
118 return fmt.Errorf("upserting repo to db: %w", err)
119 }
120
121 if !t.ks.CheckIfSubscribed(record.Knot) {
122 if err := t.ks.SubscribeHost(ctx, record.Knot, !t.cfg.KnotUseSSL); err != nil {
123 return fmt.Errorf("subscribing to knot: %w", err)
124 }
125 }
126
127 case tapc.RecordDeleteAction:
128 if err := db.DeleteRepo(ctx, t.db, evt.Did, evt.Rkey); err != nil {
129 return fmt.Errorf("deleting repo from db: %w", err)
130 }
131 }
132 return nil
133}
134
135// isPrivate checks if host is private network. It doesn't perform DNS resolution
136func isPrivate(host string) bool {
137 if host == "localhost" {
138 return true
139 }
140 addr, err := netip.ParseAddr(host)
141 if err != nil {
142 return false
143 }
144 return isPrivateAddr(addr)
145}
146
147func isPrivateAddr(addr netip.Addr) bool {
148 return addr.IsLoopback() ||
149 addr.IsPrivate() ||
150 addr.IsLinkLocalUnicast() ||
151 addr.IsLinkLocalMulticast()
152}