becca.monster
forked from
tangled.org/core
Monorepo for Tangled
wip: knotmirror/xrpc: `sync.listRepos` & `sync.listHosts`
Signed-off-by: Seongmin Lee <git@boltless.me>
author
Seongmin Lee
date
(Jul 16, 2026, 5:18 AM +0900)
commit
765a89f3
765a89f3cadd54e4d351f6dfbddb3b12ecb70da4
parent
ca808bc4
ca808bc423c3f5d7fd52aea2ea2cc30df6902d74
change-id
vuwsqpkq
vuwsqpkqyuotuorrkytspmxpwxvstkrt
+252
+39
knotmirror/db/hosts.go
+39
knotmirror/db/hosts.go
···
7
7
"fmt"
8
8
"log"
9
9
10
+
"tangled.org/core/appview/pagination"
10
11
"tangled.org/core/knotmirror/models"
11
12
)
12
13
···
68
69
}
69
70
}
70
71
return tx.Commit()
72
+
}
73
+
74
+
func ListHostsPaginated(ctx context.Context, e DBTX, page pagination.Page) ([]models.Host, error) {
75
+
var args []any
76
+
77
+
pageClause := ""
78
+
if page.Limit > 0 {
79
+
pageClause = " limit $1 offset $2 "
80
+
args = append(args, page.Limit, page.Offset)
81
+
}
82
+
83
+
rows, err := e.QueryContext(ctx,
84
+
`select hostname, no_ssl, status, last_seq
85
+
from hosts` + pageClause,
86
+
args...,
87
+
)
88
+
if err != nil {
89
+
return nil, err
90
+
}
91
+
defer rows.Close()
92
+
93
+
var hosts []models.Host
94
+
for rows.Next() {
95
+
var host models.Host
96
+
if err := rows.Scan(
97
+
&host.Hostname,
98
+
&host.NoSSL,
99
+
&host.Status,
100
+
&host.LastSeq,
101
+
); err != nil {
102
+
return nil, fmt.Errorf("scanning row: %w", err)
103
+
}
104
+
hosts = append(hosts, host)
105
+
}
106
+
if err := rows.Err(); err != nil {
107
+
return nil, fmt.Errorf("scanning rows: %w ", err)
108
+
}
109
+
return hosts, nil
71
110
}
72
111
73
112
func ListHosts(ctx context.Context, e DBTX, status models.HostStatus) ([]models.Host, error) {
+48
knotmirror/models/models.go
+48
knotmirror/models/models.go
···
22
22
ErrorMsg string
23
23
RetryCount int
24
24
RetryAfter int64 // Unix timestamp (seconds)
25
+
26
+
// knot-relay data
27
+
Status GitRepoStatus
28
+
UpstreamStatus GitRepoStatus
29
+
Digest string
25
30
}
26
31
27
32
func (r *Repo) AtUri() syntax.ATURI {
···
31
36
func (r *Repo) RepoIdentifier() string {
32
37
return r.RepoDid.String()
33
38
}
39
+
40
+
func (r *Repo) IsActive() bool {
41
+
return true // TODO: remove this hard-coded line
42
+
return (r.Status == GitRepoStatusActive || r.Status == GitRepoStatusThrottled) && r.UpstreamStatus == GitRepoStatusActive
43
+
}
44
+
45
+
func (a *Repo) GitRepoStatus() GitRepoStatus {
46
+
if a.Status != GitRepoStatusActive {
47
+
if a.Status == GitRepoStatusHostThrottled {
48
+
return GitRepoStatusThrottled
49
+
}
50
+
return a.Status
51
+
}
52
+
return a.UpstreamStatus
53
+
}
54
+
55
+
// Returns a pointer to a copy of status string; or nil if status is active.
56
+
//
57
+
// Helpful for account info responses which have a boolean 'active' and optional 'status' field (like the #account message)
58
+
func (a *Repo) StatusField() *string {
59
+
if a.IsActive() {
60
+
return nil
61
+
}
62
+
s := string(a.GitRepoStatus())
63
+
return &s
64
+
}
65
+
66
+
type GitRepoStatus string
67
+
68
+
const (
69
+
// GitRepoStatusActive is not in the spec but used internally
70
+
GitRepoStatusActive GitRepoStatus = "active"
71
+
72
+
GitRepoStatusDeactivated GitRepoStatus = "deactivated"
73
+
GitRepoStatusDeleted GitRepoStatus = "deleted"
74
+
GitRepoStatusDesynchronized GitRepoStatus = "desynchronized"
75
+
GitRepoStatusSuspended GitRepoStatus = "suspended"
76
+
GitRepoStatusTakendown GitRepoStatus = "takendown"
77
+
GitRepoStatusThrottled GitRepoStatus = "throttled"
78
+
79
+
// generic "not active, but not known" status
80
+
GitRepoStatusHostThrottled GitRepoStatus = "host-throttled"
81
+
)
34
82
35
83
type RepoState string
36
84
+78
knotmirror/xrpc/sync_list_hosts.go
+78
knotmirror/xrpc/sync_list_hosts.go
···
1
+
package xrpc
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
"net/http"
7
+
"strconv"
8
+
9
+
"github.com/bluesky-social/indigo/atproto/atclient"
10
+
"tangled.org/core/api/tangled"
11
+
"tangled.org/core/appview/pagination"
12
+
"tangled.org/core/knotmirror/db"
13
+
)
14
+
15
+
func (x *Xrpc) ListHosts(w http.ResponseWriter, r *http.Request) {
16
+
var (
17
+
limitQuery = r.URL.Query().Get("limit")
18
+
cursorQuery = r.URL.Query().Get("cursor")
19
+
)
20
+
l := x.logger.With("method", "sync.listHosts")
21
+
l.Debug("request")
22
+
23
+
limit := 500
24
+
if limitQuery != "" {
25
+
limit, err := strconv.Atoi(limitQuery)
26
+
if err != nil || limit < 1 || limit > 1000 {
27
+
writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("limit parameter invalid: %s", limitQuery)})
28
+
return
29
+
}
30
+
}
31
+
32
+
var cursor int64
33
+
if cursorQuery != "" {
34
+
cursor, err := strconv.ParseInt(cursorQuery, 10, 64)
35
+
if err != nil || cursor < 0 {
36
+
writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("cursor parameter invalid: %s", cursorQuery)})
37
+
return
38
+
}
39
+
}
40
+
41
+
ctx := r.Context()
42
+
43
+
out, err := x.listHosts(ctx, limit, cursor)
44
+
if err != nil {
45
+
writeJson(w, http.StatusInternalServerError, err)
46
+
return
47
+
}
48
+
writeJson(w, http.StatusOK, out)
49
+
}
50
+
51
+
func (x *Xrpc) listHosts(ctx context.Context, limit int, cursor int64) (*tangled.SyncListHosts_Output, *atclient.ErrorBody) {
52
+
page := pagination.Page{
53
+
Offset: int(cursor),
54
+
Limit: limit,
55
+
}
56
+
hosts, err := db.ListHostsPaginated(ctx, x.db, page)
57
+
if err != nil {
58
+
return nil, &atclient.ErrorBody{Name: "InternalServerError", Message: "failed to list hosts"}
59
+
}
60
+
61
+
outHosts := make([]*tangled.SyncListHosts_Host, len(hosts))
62
+
for i, host := range hosts {
63
+
outHosts[i] = &tangled.SyncListHosts_Host{
64
+
Hostname: host.Hostname,
65
+
}
66
+
}
67
+
68
+
var nextCursor *string
69
+
if len(hosts) == limit {
70
+
nextCursor = new(string)
71
+
*nextCursor = strconv.FormatInt(cursor + int64(len(hosts)), 10)
72
+
}
73
+
74
+
return &tangled.SyncListHosts_Output{
75
+
// Cursor: nextCursor,
76
+
Hosts: outHosts,
77
+
}, nil
78
+
}
+81
knotmirror/xrpc/sync_list_repos.go
+81
knotmirror/xrpc/sync_list_repos.go
···
1
+
package xrpc
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
"net/http"
7
+
"strconv"
8
+
9
+
"github.com/bluesky-social/indigo/atproto/atclient"
10
+
"tangled.org/core/api/tangled"
11
+
"tangled.org/core/appview/pagination"
12
+
"tangled.org/core/knotmirror/db"
13
+
)
14
+
15
+
func (x *Xrpc) ListRepos(w http.ResponseWriter, r *http.Request) {
16
+
var (
17
+
limitQuery = r.URL.Query().Get("limit")
18
+
cursorQuery = r.URL.Query().Get("cursor")
19
+
)
20
+
l := x.logger.With("method", "sync.listRepos")
21
+
l.Debug("request")
22
+
23
+
limit := 500
24
+
if limitQuery != "" {
25
+
limit, err := strconv.Atoi(limitQuery)
26
+
if err != nil || limit < 1 || limit > 1000 {
27
+
writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("limit parameter invalid: %s", limitQuery)})
28
+
return
29
+
}
30
+
}
31
+
32
+
var cursor int64
33
+
if cursorQuery != "" {
34
+
cursor, err := strconv.ParseInt(cursorQuery, 10, 64)
35
+
if err != nil || cursor < 0 {
36
+
writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("cursor parameter invalid: %s", cursorQuery)})
37
+
return
38
+
}
39
+
}
40
+
41
+
ctx := r.Context()
42
+
43
+
out, err := x.listRepos(ctx, limit, cursor)
44
+
if err != nil {
45
+
writeJson(w, http.StatusInternalServerError, err)
46
+
return
47
+
}
48
+
writeJson(w, http.StatusOK, out)
49
+
}
50
+
51
+
func (x *Xrpc) listRepos(ctx context.Context, limit int, cursor int64) (*tangled.SyncListRepos_Output, *atclient.ErrorBody) {
52
+
page := pagination.Page{
53
+
Offset: int(cursor),
54
+
Limit: limit,
55
+
}
56
+
repos, err := db.ListRepos(ctx, x.db, page, "", "", "", "")
57
+
if err != nil {
58
+
return nil, &atclient.ErrorBody{Name: "InternalServerError", Message: "failed to list repos"}
59
+
}
60
+
61
+
outRepos := make([]*tangled.SyncListRepos_Repo, len(repos))
62
+
for i, repo := range repos {
63
+
active := repo.IsActive()
64
+
outRepos[i] = &tangled.SyncListRepos_Repo{
65
+
Did: repo.RepoDid.String(),
66
+
Active: &active,
67
+
Status: repo.StatusField(),
68
+
}
69
+
}
70
+
71
+
var nextCursor *string
72
+
if len(repos) == limit {
73
+
nextCursor = new(string)
74
+
*nextCursor = strconv.FormatInt(cursor + int64(len(repos)), 10)
75
+
}
76
+
77
+
return &tangled.SyncListRepos_Output{
78
+
Cursor: nextCursor,
79
+
Repos: outRepos,
80
+
}, nil
81
+
}
+6
knotmirror/xrpc/xrpc.go
+6
knotmirror/xrpc/xrpc.go
···
73
73
r.Get("/"+tangled.GitTempListCommitsNSID, x.ListCommits)
74
74
r.Get("/"+tangled.GitTempListLanguagesNSID, x.ListLanguages)
75
75
r.Get("/"+tangled.GitTempListTagsNSID, x.ListTags)
76
+
77
+
// knot-relay endpoints
76
78
r.Post("/"+tangled.SyncRequestCrawlNSID, x.RequestCrawl)
79
+
r.Get("/"+tangled.SyncListHostsNSID, x.ListHosts)
80
+
r.Get("/"+tangled.SyncListReposNSID, x.ListRepos)
81
+
// r.Get("/"+tangled.SyncGetHostStatusNSID, x.GetHostStatus)
82
+
// r.Get("/"+tangled.SyncGetRepoStatusNSID, x.GetRepoStatus)
77
83
})
78
84
79
85
return r