this repo has no description
labelmaker: support label query XRPC endpoint
+154
-14
+3
cmd/labelmaker/README.md
+3
cmd/labelmaker/README.md
···
25
25
the regular user. There is no concept of running a separate set of migrations
26
26
under more privileged database user.
27
27
28
+
For database performance with many labels, it is important that `LC_COLLATE=C`.
29
+
That is, the string sort behavior must be by byte order.
30
+
28
31
## Keyword Labeler
29
32
30
33
A trivial keyword filter labeler is included. To configure it, create a JSON
+23
-3
labeling/service.go
+23
-3
labeling/service.go
···
30
30
"github.com/labstack/echo/v4/middleware"
31
31
"github.com/whyrusleeping/go-did"
32
32
"gorm.io/gorm"
33
+
"gorm.io/gorm/clause"
33
34
)
34
35
35
36
var log = logging.Logger("labelmaker")
···
61
62
func NewServer(db *gorm.DB, cs *carstore.CarStore, repoUser RepoConfig, plcURL, blobPdsURL string, useWss bool) (*Server, error) {
62
63
63
64
db.AutoMigrate(models.PDS{})
65
+
db.AutoMigrate(models.Label{})
64
66
65
67
didr := &api.PLCServer{Host: plcURL}
66
68
kmgr := indexer.NewKeyManager(didr, repoUser.SigningKey)
···
334
336
return err
335
337
}
336
338
337
-
now := time.Now().Format(util.ISO8601)
339
+
now := time.Now()
340
+
nowStr := now.Format(util.ISO8601)
338
341
labels := []label.Label{}
339
342
340
343
for _, op := range evt.RepoCommit.Ops {
···
362
365
Src: s.user.Did,
363
366
Uri: "at://" + evt.RepoCommit.Repo,
364
367
Val: val,
365
-
Cts: now,
368
+
Cts: nowStr,
366
369
})
367
370
} else {
368
371
labels = append(labels, label.Label{
···
370
373
Uri: uri,
371
374
Cid: &cidStr,
372
375
Val: val,
373
-
Cts: now,
376
+
Cts: nowStr,
374
377
})
375
378
}
376
379
}
···
384
387
}
385
388
labeluri := "at://" + s.user.Did + "/" + path
386
389
log.Infof("persisted label: %s", labeluri)
390
+
}
391
+
392
+
// ... and database ...
393
+
var labelRows []models.Label
394
+
for _, l := range labels {
395
+
lr := models.Label{
396
+
Uri: l.Uri,
397
+
SourceDid: l.Src,
398
+
Cid: l.Cid,
399
+
Val: l.Val,
400
+
CreatedAt: now,
401
+
}
402
+
labelRows = append(labelRows, lr)
403
+
}
404
+
if len(labelRows) > 0 {
405
+
// TODO(bnewbold): don't clobber action labels (aka, human interventions)
406
+
s.db.Clauses(clause.OnConflict{DoNothing: true}).Create(&labelRows)
387
407
}
388
408
389
409
// ... then re-publish as XRPCStreamEvent
+35
-10
labeling/xrpc_endpoints.go
+35
-10
labeling/xrpc_endpoints.go
···
4
4
"strconv"
5
5
6
6
atproto "github.com/bluesky-social/indigo/api/atproto"
7
+
label "github.com/bluesky-social/indigo/api/label"
7
8
8
9
"github.com/labstack/echo/v4"
9
10
"go.opentelemetry.io/otel"
10
11
)
11
12
12
13
func (s *Server) RegisterHandlersComAtproto(e *echo.Echo) error {
14
+
// handle/account hosting
13
15
e.GET("/xrpc/com.atproto.identity.resolveHandle", s.HandleComAtprotoIdentityResolveHandle)
14
-
e.GET("/xrpc/com.atproto.repo.describeRepo", s.HandleComAtprotoRepoDescribeRepo)
15
-
e.GET("/xrpc/com.atproto.repo.getRecord", s.HandleComAtprotoRepoGetRecord)
16
-
e.GET("/xrpc/com.atproto.repo.listRecords", s.HandleComAtprotoRepoListRecords)
17
16
e.GET("/xrpc/com.atproto.server.describeServer", s.HandleComAtprotoServerDescribeServer)
18
-
e.GET("/xrpc/com.atproto.sync.getHead", s.HandleComAtprotoSyncGetHead)
17
+
// TODO: session create/refresh/delete?
18
+
//e.GET("/xrpc/com.atproto.account.get", s.HandleComAtprotoAccountGet)
19
+
20
+
// label-specific
21
+
e.GET("/xrpc/com.atproto.label.query", s.HandleComAtprotoLabelQuery)
22
+
23
+
// minimal moderation reporting/actioning
24
+
// TODO: com.atproto.admin.takeModerationAction,reverseModerationAction,getModerationAction(s)
25
+
// TODO: com.atproto.report.create, com.atproto.admin.getModerationReport(s)
26
+
// TODO: com.atproto.admin.resolveModerationReports
27
+
// TODO: proxy all the rest to BGS?
28
+
19
29
return nil
20
30
}
21
31
···
114
124
return c.JSON(200, out)
115
125
}
116
126
117
-
func (s *Server) HandleComAtprotoSyncGetHead(c echo.Context) error {
118
-
ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoSyncGetHead")
127
+
func (s *Server) HandleComAtprotoLabelQuery(c echo.Context) error {
128
+
ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoLabelQuery")
119
129
defer span.End()
120
-
did := c.QueryParam("did")
121
-
var out *atproto.SyncGetHead_Output
130
+
cursor := c.QueryParam("cursor")
131
+
132
+
var limit int
133
+
if p := c.QueryParam("limit"); p != "" {
134
+
var err error
135
+
limit, err = strconv.Atoi(p)
136
+
if err != nil {
137
+
return err
138
+
}
139
+
} else {
140
+
limit = 20
141
+
}
142
+
143
+
sources := c.QueryParams()["sources"]
144
+
145
+
uriPatterns := c.QueryParams()["uriPatterns"]
146
+
var out *label.Query_Output
122
147
var handleErr error
123
-
// func (s *Server) handleComAtprotoSyncGetHead(ctx context.Context,did string) (*atproto.SyncGetHead_Output, error)
124
-
out, handleErr = s.handleComAtprotoSyncGetHead(ctx, did)
148
+
// func (s *Server) handleComAtprotoLabelQuery(ctx context.Context,cursor string,limit int,sources []string,uriPatterns []string) (*comatprototypes.LabelQuery_Output, error)
149
+
out, handleErr = s.handleComAtprotoLabelQuery(ctx, cursor, limit, sources, uriPatterns)
125
150
if handleErr != nil {
126
151
return handleErr
127
152
}
+79
-1
labeling/xrpc_handlers.go
+79
-1
labeling/xrpc_handlers.go
···
5
5
"context"
6
6
"fmt"
7
7
"io"
8
+
"strconv"
9
+
"strings"
8
10
9
11
atproto "github.com/bluesky-social/indigo/api/atproto"
12
+
label "github.com/bluesky-social/indigo/api/label"
10
13
lexutil "github.com/bluesky-social/indigo/lex/util"
11
-
util "github.com/bluesky-social/indigo/util"
14
+
"github.com/bluesky-social/indigo/models"
15
+
"github.com/bluesky-social/indigo/util"
12
16
13
17
"github.com/ipfs/go-cid"
14
18
)
···
102
106
func (s *Server) handleComAtprotoSyncGetHead(ctx context.Context, did string) (*atproto.SyncGetHead_Output, error) {
103
107
panic("not yet implemented")
104
108
}
109
+
110
+
func (s *Server) handleComAtprotoLabelQuery(ctx context.Context, cursor string, limit int, sources, uriPatterns []string) (*label.Query_Output, error) {
111
+
112
+
if limit <= 0 {
113
+
limit = 20
114
+
}
115
+
if limit > 100 {
116
+
limit = 100
117
+
}
118
+
119
+
q := s.db.Limit(limit).Order("id desc")
120
+
if cursor != "" {
121
+
cursorID, err := strconv.Atoi(cursor)
122
+
if err != nil {
123
+
return nil, err
124
+
}
125
+
q = q.Where("id < ?", cursorID)
126
+
}
127
+
128
+
srcQuery := s.db
129
+
fmt.Printf("%v\n", sources)
130
+
for _, src := range sources {
131
+
if src == "*" {
132
+
continue
133
+
}
134
+
srcQuery = srcQuery.Or("source_did = ?", src)
135
+
}
136
+
if srcQuery != s.db {
137
+
q = q.Where(srcQuery)
138
+
}
139
+
140
+
uriQuery := s.db
141
+
for _, pat := range uriPatterns {
142
+
if strings.HasSuffix(pat, "*") {
143
+
likePat := []rune(pat)
144
+
likePat[len(likePat)-1] = '%'
145
+
uriQuery = uriQuery.Or("uri LIKE ?", string(likePat))
146
+
} else {
147
+
uriQuery = uriQuery.Or("uri = ?", pat)
148
+
}
149
+
}
150
+
if uriQuery != s.db {
151
+
q = q.Where(uriQuery)
152
+
}
153
+
154
+
var labelRows []models.Label
155
+
result := q.Find(&labelRows)
156
+
if result.Error != nil {
157
+
return nil, result.Error
158
+
}
159
+
160
+
var nextCursor string
161
+
if len(labelRows) >= 1 && len(labelRows) == limit {
162
+
nextCursor = strconv.FormatUint(labelRows[len(labelRows)-1].ID, 10)
163
+
}
164
+
165
+
labelObjs := []*label.Label{}
166
+
for _, row := range labelRows {
167
+
labelObjs = append(labelObjs, &label.Label{
168
+
Src: row.SourceDid,
169
+
Uri: row.Uri,
170
+
Cid: row.Cid,
171
+
Val: row.Val,
172
+
Cts: row.CreatedAt.Format(util.ISO8601),
173
+
})
174
+
}
175
+
out := label.Query_Output{
176
+
Labels: labelObjs,
177
+
}
178
+
if nextCursor != "" {
179
+
out.Cursor = &nextCursor
180
+
}
181
+
return &out, nil
182
+
}
+14
models/models.go
+14
models/models.go
···
121
121
Host: "http://" + pds.Host,
122
122
}
123
123
}
124
+
125
+
// The CreatedAt column corresponds to the 'cat' timestamp on label records. The UpdatedAt column is database-specific.
126
+
//
127
+
// NOTE: to get fast string-prefix queries on Uri via the idx_uri_src_val_cid index, it is important that the PostgreSQL LC_COLLATE="C"
128
+
type Label struct {
129
+
ID uint64 `gorm:"primaryKey"`
130
+
Uri string `gorm:"uniqueIndex:idx_uri_src_val_cid;not null"`
131
+
SourceDid string `gorm:"uniqueIndex:idx_uri_src_val_cid;uniqueIndex:idx_src_rkey;not null"`
132
+
Val string `gorm:"uniqueIndex:idx_uri_src_val_cid;not null"`
133
+
Cid *string `gorm:"uniqueIndex:idx_uri_src_val_cid"`
134
+
RepoRKey *string `gorm:"uniqueIndex:idx_src_rkey"`
135
+
CreatedAt time.Time
136
+
UpdatedAt time.Time
137
+
}