forked from
tangled.org/core
Mirror of @tangled.org/core. Running on a Raspberry Pi Zero 2 (Please be gentle).
spindle: ingest repo records in spindle
allows the spindle to dynamically configure the knots it is listening
to.
Signed-off-by: oppiliappan <me@oppi.li>
+90
-2
+10
spindle/db/db.go
+10
spindle/db/db.go
···
35
35
did text primary key
36
36
);
37
37
38
+
create table if not exists repos (
39
+
id integer primary key autoincrement,
40
+
knot text not null,
41
+
owner text not null,
42
+
name text not null,
43
+
addedAt text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
44
+
45
+
unique(owner, name)
46
+
);
47
+
38
48
-- status event for a single workflow
39
49
create table if not exists events (
40
50
rkey text not null,
+28
spindle/db/repos.go
+28
spindle/db/repos.go
···
1
+
package db
2
+
3
+
func (d *DB) AddRepo(knot, owner, name string) error {
4
+
_, err := d.Exec(`insert or ignore into repos (knot, owner, name) values (?, ?, ?)`, knot, owner, name)
5
+
return err
6
+
}
7
+
8
+
func (d *DB) Knots() ([]string, error) {
9
+
rows, err := d.Query(`select knot from repos`)
10
+
if err != nil {
11
+
return nil, err
12
+
}
13
+
14
+
var knots []string
15
+
for rows.Next() {
16
+
var knot string
17
+
if err := rows.Scan(&knot); err != nil {
18
+
return nil, err
19
+
}
20
+
knots = append(knots, knot)
21
+
}
22
+
23
+
if err = rows.Err(); err != nil {
24
+
return nil, err
25
+
}
26
+
27
+
return knots, nil
28
+
}
+52
-2
spindle/ingester.go
+52
-2
spindle/ingester.go
···
5
5
"encoding/json"
6
6
"fmt"
7
7
8
-
"github.com/bluesky-social/jetstream/pkg/models"
9
8
"tangled.sh/tangled.sh/core/api/tangled"
9
+
"tangled.sh/tangled.sh/core/knotclient"
10
+
11
+
"github.com/bluesky-social/jetstream/pkg/models"
10
12
)
11
13
12
14
type Ingester func(ctx context.Context, e *models.Event) error
···
31
29
switch e.Commit.Collection {
32
30
case tangled.SpindleMemberNSID:
33
31
s.ingestMember(ctx, e)
32
+
case tangled.RepoNSID:
33
+
s.ingestRepo(ctx, e)
34
34
}
35
35
36
36
return err
···
72
68
return fmt.Errorf("failed to enforce permissions: %w", err)
73
69
}
74
70
75
-
if err := s.e.AddMember(rbacDomain, record.Subject); err != nil {
71
+
if err := s.e.AddKnotMember(rbacDomain, record.Subject); err != nil {
76
72
l.Error("failed to add member", "error", err)
77
73
return fmt.Errorf("failed to add member: %w", err)
78
74
}
···
83
79
return fmt.Errorf("failed to add did: %w", err)
84
80
}
85
81
s.jc.AddDid(did)
82
+
83
+
return nil
84
+
85
+
}
86
+
return nil
87
+
}
88
+
89
+
func (s *Spindle) ingestRepo(_ context.Context, e *models.Event) error {
90
+
var err error
91
+
92
+
l := s.l.With("component", "ingester", "record", tangled.RepoNSID)
93
+
94
+
switch e.Commit.Operation {
95
+
case models.CommitOperationCreate, models.CommitOperationUpdate:
96
+
raw := e.Commit.Record
97
+
record := tangled.Repo{}
98
+
err = json.Unmarshal(raw, &record)
99
+
if err != nil {
100
+
l.Error("invalid record", "error", err)
101
+
return err
102
+
}
103
+
104
+
domain := s.cfg.Server.Hostname
105
+
if s.cfg.Server.Dev {
106
+
domain = s.cfg.Server.ListenAddr
107
+
}
108
+
109
+
// no spindle configured for this repo
110
+
if record.Spindle == nil {
111
+
return nil
112
+
}
113
+
114
+
// this repo did not want this spindle
115
+
if *record.Spindle != domain {
116
+
return nil
117
+
}
118
+
119
+
// add this repo to the watch list
120
+
if err := s.db.AddRepo(record.Knot, record.Owner, record.Name); err != nil {
121
+
l.Error("failed to add repo", "error", err)
122
+
return fmt.Errorf("failed to add repo: %w", err)
123
+
}
124
+
125
+
// add this knot to the event consumer
126
+
src := knotclient.NewEventSource(record.Knot)
127
+
s.ks.AddSource(context.Background(), src)
86
128
87
129
return nil
88
130