alice.pds.demo.boltless.dev
This repository has no description
spindle: improve member ingestion
Signed-off-by: oppiliappan <me@oppi.li>
author
oppiliappan
date
(Aug 8, 2025, 5:12 PM +0100)
commit
23776d5e
23776d5eca3824f2410a9c9a4fcf64a07f0fa65d
parent
4e03f297
4e03f2978a3185589848093270cb3b54fb6d53f0
change-id
vokloxlu
vokloxluvoxxwoquqlmkvxpmzsswtvok
+121
-8
+4
appview/ingester.go
+4
appview/ingester.go
···
387
387
if err != nil {
388
388
return fmt.Errorf("failed to update ACLs: %w", err)
389
389
}
390
+
391
+
l.Info("added spindle member")
390
392
case models.CommitOperationDelete:
391
393
rkey := e.Commit.RKey
392
394
···
433
435
if err = i.Enforcer.E.SavePolicy(); err != nil {
434
436
return fmt.Errorf("failed to save ACLs: %w", err)
435
437
}
438
+
439
+
l.Info("removed spindle member")
436
440
}
437
441
438
442
return nil
+4
-4
appview/spindles/spindles.go
+4
-4
appview/spindles/spindles.go
···
619
619
620
620
if string(spindles[0].Owner) != user.Did {
621
621
l.Error("unauthorized", "user", user.Did, "owner", spindles[0].Owner)
622
-
s.Pages.Notice(w, noticeId, "Failed to add member, unauthorized attempt.")
622
+
s.Pages.Notice(w, noticeId, "Failed to remove member, unauthorized attempt.")
623
623
return
624
624
}
625
625
626
626
member := r.FormValue("member")
627
627
if member == "" {
628
628
l.Error("empty member")
629
-
s.Pages.Notice(w, noticeId, "Failed to add member, empty form.")
629
+
s.Pages.Notice(w, noticeId, "Failed to remove member, empty form.")
630
630
return
631
631
}
632
632
l = l.With("member", member)
···
634
634
memberId, err := s.IdResolver.ResolveIdent(r.Context(), member)
635
635
if err != nil {
636
636
l.Error("failed to resolve member identity to handle", "err", err)
637
-
s.Pages.Notice(w, noticeId, "Failed to add member, identity resolution failed.")
637
+
s.Pages.Notice(w, noticeId, "Failed to remove member, identity resolution failed.")
638
638
return
639
639
}
640
640
if memberId.Handle.IsInvalidHandle() {
641
641
l.Error("failed to resolve member identity to handle")
642
-
s.Pages.Notice(w, noticeId, "Failed to add member, identity resolution failed.")
642
+
s.Pages.Notice(w, noticeId, "Failed to remove member, identity resolution failed.")
643
643
return
644
644
}
645
645
+15
spindle/db/db.go
+15
spindle/db/db.go
···
45
45
unique(owner, name)
46
46
);
47
47
48
+
create table if not exists spindle_members (
49
+
-- identifiers for the record
50
+
id integer primary key autoincrement,
51
+
did text not null,
52
+
rkey text not null,
53
+
54
+
-- data
55
+
instance text not null,
56
+
subject text not null,
57
+
created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
58
+
59
+
-- constraints
60
+
unique (did, instance, subject)
61
+
);
62
+
48
63
-- status event for a single workflow
49
64
create table if not exists events (
50
65
rkey text not null,
+59
spindle/db/member.go
+59
spindle/db/member.go
···
1
+
package db
2
+
3
+
import (
4
+
"time"
5
+
6
+
"github.com/bluesky-social/indigo/atproto/syntax"
7
+
)
8
+
9
+
type SpindleMember struct {
10
+
Id int
11
+
Did syntax.DID // owner of the record
12
+
Rkey string // rkey of the record
13
+
Instance string
14
+
Subject syntax.DID // the member being added
15
+
Created time.Time
16
+
}
17
+
18
+
func AddSpindleMember(db *DB, member SpindleMember) error {
19
+
_, err := db.Exec(
20
+
`insert or ignore into spindle_members (did, rkey, instance, subject) values (?, ?, ?, ?)`,
21
+
member.Did,
22
+
member.Rkey,
23
+
member.Instance,
24
+
member.Subject,
25
+
)
26
+
return err
27
+
}
28
+
29
+
func RemoveSpindleMember(db *DB, owner_did, rkey string) error {
30
+
_, err := db.Exec(
31
+
"delete from spindle_members where did = ? and rkey = ?",
32
+
owner_did,
33
+
rkey,
34
+
)
35
+
return err
36
+
}
37
+
38
+
func GetSpindleMember(db *DB, did, rkey string) (*SpindleMember, error) {
39
+
query :=
40
+
`select id, did, rkey, instance, subject, created
41
+
from spindle_members
42
+
where did = ? and rkey = ?`
43
+
44
+
var member SpindleMember
45
+
var createdAt string
46
+
err := db.QueryRow(query, did, rkey).Scan(
47
+
&member.Id,
48
+
&member.Did,
49
+
&member.Rkey,
50
+
&member.Instance,
51
+
&member.Subject,
52
+
&createdAt,
53
+
)
54
+
if err != nil {
55
+
return nil, err
56
+
}
57
+
58
+
return &member, nil
59
+
}
+39
-4
spindle/ingester.go
+39
-4
spindle/ingester.go
···
5
5
"encoding/json"
6
6
"errors"
7
7
"fmt"
8
+
"time"
8
9
9
10
"tangled.sh/tangled.sh/core/api/tangled"
10
11
"tangled.sh/tangled.sh/core/eventconsumer"
11
12
"tangled.sh/tangled.sh/core/idresolver"
12
13
"tangled.sh/tangled.sh/core/rbac"
14
+
"tangled.sh/tangled.sh/core/spindle/db"
13
15
14
16
comatproto "github.com/bluesky-social/indigo/api/atproto"
15
17
"github.com/bluesky-social/indigo/atproto/identity"
···
50
52
}
51
53
52
54
func (s *Spindle) ingestMember(_ context.Context, e *models.Event) error {
55
+
var err error
53
56
did := e.Did
54
-
var err error
57
+
rkey := e.Commit.RKey
55
58
56
59
l := s.l.With("component", "ingester", "record", tangled.SpindleMemberNSID)
57
60
···
66
69
}
67
70
68
71
domain := s.cfg.Server.Hostname
69
-
if s.cfg.Server.Dev {
70
-
domain = s.cfg.Server.ListenAddr
71
-
}
72
72
recordInstance := record.Instance
73
73
74
74
if recordInstance != domain {
···
82
82
return fmt.Errorf("failed to enforce permissions: %w", err)
83
83
}
84
84
85
+
if err := db.AddSpindleMember(s.db, db.SpindleMember{
86
+
Did: syntax.DID(did),
87
+
Rkey: rkey,
88
+
Instance: recordInstance,
89
+
Subject: syntax.DID(record.Subject),
90
+
Created: time.Now(),
91
+
}); err != nil {
92
+
l.Error("failed to add member", "error", err)
93
+
return fmt.Errorf("failed to add member: %w", err)
94
+
}
95
+
85
96
if err := s.e.AddSpindleMember(rbacDomain, record.Subject); err != nil {
86
97
l.Error("failed to add member", "error", err)
87
98
return fmt.Errorf("failed to add member: %w", err)
···
95
106
s.jc.AddDid(record.Subject)
96
107
97
108
return nil
109
+
110
+
case models.CommitOperationDelete:
111
+
record, err := db.GetSpindleMember(s.db, did, rkey)
112
+
if err != nil {
113
+
l.Error("failed to find member", "error", err)
114
+
return fmt.Errorf("failed to find member: %w", err)
115
+
}
116
+
117
+
if err := db.RemoveSpindleMember(s.db, did, rkey); err != nil {
118
+
l.Error("failed to remove member", "error", err)
119
+
return fmt.Errorf("failed to remove member: %w", err)
120
+
}
121
+
122
+
if err := s.e.RemoveSpindleMember(rbacDomain, record.Subject.String()); err != nil {
123
+
l.Error("failed to add member", "error", err)
124
+
return fmt.Errorf("failed to add member: %w", err)
125
+
}
126
+
l.Info("added member from firehose", "member", record.Subject)
127
+
128
+
if err := s.db.RemoveDid(record.Subject.String()); err != nil {
129
+
l.Error("failed to add did", "error", err)
130
+
return fmt.Errorf("failed to add did: %w", err)
131
+
}
132
+
s.jc.RemoveDid(record.Subject.String())
98
133
99
134
}
100
135
return nil