an app.bsky.* indexer

Compare changes

Choose any two refs to compare.

-80
cmd/backfiller/models.go
···
-
package main
-
-
import (
-
"bytes"
-
"log/slog"
-
-
appbsky "github.com/bluesky-social/indigo/api/bsky"
-
)
-
-
type Account struct{}
-
-
// func NewAccount
-
-
type Profile struct {
-
ID uint `gorm:"primaryKey"`
-
AtUri string
-
DisplayName *string
-
}
-
-
func NewProfile(rec []byte) *Profile {
-
var out appbsky.ActorProfile
-
if err := out.UnmarshalCBOR(bytes.NewReader(rec)); err != nil {
-
slog.Error("could not unmarshal profile CBOR", "err", err)
-
return nil
-
}
-
return &Profile{
-
DisplayName: out.DisplayName,
-
}
-
}
-
-
type List struct{}
-
-
// func NewList
-
-
type Labeler struct{}
-
-
// func NewFeedGenerator
-
-
type FeedGenerator struct {
-
ID uint `gorm:"primaryKey"`
-
AtUri string
-
DisplayName string
-
}
-
-
func NewFeedGenerator(rec []byte) *FeedGenerator {
-
var out appbsky.FeedGenerator
-
if err := out.UnmarshalCBOR(bytes.NewReader(rec)); err != nil {
-
slog.Error("could not unmarshal feedgen CBOR", "err", err)
-
return nil
-
}
-
return &FeedGenerator{
-
DisplayName: out.DisplayName,
-
}
-
}
-
-
type Post struct {
-
ID uint `gorm:"primaryKey"`
-
AtUri string
-
Text string
-
}
-
-
func NewPost(rec []byte) *Post {
-
return nil
-
}
-
-
type Like struct{}
-
-
// func NewLike
-
-
type StarterPack struct{}
-
-
// func NewStarterPack
-
-
type Verification struct{}
-
-
// func NewVerification
-
-
type Lexicon struct{}
-
-
// func NewLexicon
+6
models/strongref.go
···
+
package models
+
+
type StrongRef struct {
+
Uri string
+
Cid string
+
}
+87
models/feed_threadgate.go
···
+
package models
+
+
import (
+
"bytes"
+
"log/slog"
+
"time"
+
+
appbsky "github.com/bluesky-social/indigo/api/bsky"
+
"github.com/bluesky-social/indigo/atproto/syntax"
+
)
+
+
type FeedThreadgate struct {
+
ID string `gorm:"primaryKey"`
+
+
AllowRules []FeedThreadgate_AllowRule
+
CreatedAt string
+
HiddenReplies []FeedThreadgate_HiddenReply
+
Post string
+
+
AutoCreatedAt time.Time `gorm:"autoCreateTime"`
+
AutoUpdatedAt time.Time `gorm:"autoUpdateTime"`
+
}
+
+
type FeedThreadgate_AllowRule struct {
+
FeedThreadgateID string
+
Rule string
+
}
+
+
type FeedThreadgate_HiddenReply struct {
+
FeedThreadgateID string
+
Uri string
+
}
+
+
func NewFeedThreadgate(uri syntax.ATURI, rec []byte) *FeedThreadgate {
+
var out appbsky.FeedThreadgate
+
if err := out.UnmarshalCBOR(bytes.NewReader(rec)); err != nil {
+
slog.Error("could not unmarshal feed threadgate CBOR", "err", err)
+
return nil
+
}
+
+
threadgate := FeedThreadgate{
+
ID: string(uri),
+
CreatedAt: out.CreatedAt,
+
Post: out.Post,
+
}
+
+
for _, hidden := range out.HiddenReplies {
+
threadgate.HiddenReplies = append(threadgate.HiddenReplies, FeedThreadgate_HiddenReply{
+
FeedThreadgateID: threadgate.ID,
+
Uri: hidden,
+
})
+
}
+
+
if out.Allow != nil {
+
for _, rule := range out.Allow {
+
if rule.FeedThreadgate_MentionRule != nil {
+
threadgate.AllowRules = append(threadgate.AllowRules, FeedThreadgate_AllowRule{
+
FeedThreadgateID: threadgate.ID,
+
Rule: rule.FeedThreadgate_MentionRule.LexiconTypeID,
+
})
+
}
+
+
if rule.FeedThreadgate_FollowerRule != nil {
+
threadgate.AllowRules = append(threadgate.AllowRules, FeedThreadgate_AllowRule{
+
FeedThreadgateID: threadgate.ID,
+
Rule: rule.FeedThreadgate_FollowerRule.LexiconTypeID,
+
})
+
}
+
+
if rule.FeedThreadgate_FollowingRule != nil {
+
threadgate.AllowRules = append(threadgate.AllowRules, FeedThreadgate_AllowRule{
+
FeedThreadgateID: threadgate.ID,
+
Rule: rule.FeedThreadgate_FollowingRule.LexiconTypeID,
+
})
+
}
+
+
if rule.FeedThreadgate_ListRule != nil {
+
threadgate.AllowRules = append(threadgate.AllowRules, FeedThreadgate_AllowRule{
+
FeedThreadgateID: threadgate.ID,
+
Rule: rule.FeedThreadgate_ListRule.LexiconTypeID,
+
})
+
}
+
}
+
}
+
+
return &threadgate
+
}
-20
models/models.go
···
-
package models
-
-
// - [X] ActorProfile *
-
// - [X] ActorStatus *
-
// - [X] FeedGenerator *
-
// - [X] FeedLike
-
// - [X] FeedPost
-
// - [X] FeedPostgate
-
// - [X] FeedRepost
-
// - [X] FeedThreadgate
-
// - [X] GraphBlock *
-
// - [X] GraphFollow
-
// - [X] GraphList *
-
// - [X] GraphListblock *
-
// - [X] GraphListitem *
-
// - [X] GraphStarterpack *
-
// - [X] GraphVerification *
-
// - [X] LabelerService *
-
// - [X] ActorDeclaration
-
// - [ ] LexiconSchema *
+10 -26
models/actor_profile.go
···
CreatedAt *string
Description *string
DisplayName *string
-
JoinedViaStarterPack []ActorProfile_JoinedViaStarterPack
+
JoinedViaStarterPack *StrongRef `gorm:"embedded;embeddedPrefix:starterpack_"`
Labels []ActorProfile_Label
-
PinnedPost []ActorProfile_PinnedPost
+
PinnedPost *StrongRef `gorm:"embedded;embeddedPrefix:pinnedpost_"`
AutoCreatedAt time.Time `gorm:"autoCreateTime"`
AutoUpdatedAt time.Time `gorm:"autoUpdateTime"`
···
Value string
}
-
type ActorProfile_JoinedViaStarterPack struct {
-
ActorProfileID string
-
StrongRef
-
}
-
-
type ActorProfile_PinnedPost struct {
-
ActorProfileID string
-
StrongRef
-
}
-
func NewActorProfile(uri syntax.ATURI, rec []byte) *ActorProfile {
var out appbsky.ActorProfile
if err := out.UnmarshalCBOR(bytes.NewReader(rec)); err != nil {
···
}
if out.JoinedViaStarterPack != nil {
-
profile.JoinedViaStarterPack = append(profile.JoinedViaStarterPack, ActorProfile_JoinedViaStarterPack{
-
ActorProfileID: profile.ID,
-
StrongRef: StrongRef{
-
Uri: out.JoinedViaStarterPack.Uri,
-
Cid: out.JoinedViaStarterPack.Cid,
-
},
-
})
+
profile.JoinedViaStarterPack = &StrongRef{
+
Uri: out.JoinedViaStarterPack.Uri,
+
Cid: out.JoinedViaStarterPack.Cid,
+
}
}
if out.PinnedPost != nil {
-
profile.PinnedPost = append(profile.PinnedPost, ActorProfile_PinnedPost{
-
ActorProfileID: profile.ID,
-
StrongRef: StrongRef{
-
Uri: out.PinnedPost.Uri,
-
Cid: out.PinnedPost.Cid,
-
},
-
})
+
profile.PinnedPost = &StrongRef{
+
Uri: out.PinnedPost.Uri,
+
Cid: out.PinnedPost.Cid,
+
}
}
if out.Labels != nil && out.Labels.LabelDefs_SelfLabels != nil && out.Labels.LabelDefs_SelfLabels.Values != nil {
+8 -24
models/feed_like.go
···
ID string `gorm:"primaryKey"`
CreatedAt string
-
Subject FeedLike_Subject
-
Via FeedLike_Via
+
Subject *StrongRef `gorm:"embedded;embeddedPrefix:subject_"`
+
Via *StrongRef `gorm:"embedded;embeddedPrefix:via_"`
AutoCreatedAt time.Time `gorm:"autoCreateTime"`
AutoUpdatedAt time.Time `gorm:"autoUpdateTime"`
}
-
type FeedLike_Subject struct {
-
FeedLikeID string
-
StrongRef
-
}
-
-
type FeedLike_Via struct {
-
FeedLikeID string
-
StrongRef
-
}
-
func NewFeedLike(uri syntax.ATURI, rec []byte) *FeedLike {
var out appbsky.FeedLike
if err := out.UnmarshalCBOR(bytes.NewReader(rec)); err != nil {
···
}
if out.Subject != nil {
-
like.Subject = FeedLike_Subject{
-
FeedLikeID: like.ID,
-
StrongRef: StrongRef{
-
Uri: out.Subject.Uri,
-
Cid: out.Subject.Cid,
-
},
+
like.Subject = &StrongRef{
+
Uri: out.Subject.Uri,
+
Cid: out.Subject.Cid,
}
}
if out.Via != nil {
-
like.Via = FeedLike_Via{
-
FeedLikeID: like.ID,
-
StrongRef: StrongRef{
-
Uri: out.Via.Uri,
-
Cid: out.Via.Cid,
-
},
+
like.Via = &StrongRef{
+
Uri: out.Via.Uri,
+
Cid: out.Via.Cid,
}
}
+2 -2
models/feed_postgate.go
···
type FeedPostgate_EmbeddingRule struct {
FeedPostgateID string
-
DisableRule string
+
Rule string
}
func NewFeedPostgate(uri syntax.ATURI, rec []byte) *FeedPostgate {
···
if rule.FeedPostgate_DisableRule != nil {
postgate.EmbeddingRules = append(postgate.EmbeddingRules, FeedPostgate_EmbeddingRule{
FeedPostgateID: postgate.ID,
-
DisableRule: rule.FeedPostgate_DisableRule.LexiconTypeID,
+
Rule: rule.FeedPostgate_DisableRule.LexiconTypeID,
})
}
}
+1
models/graph_follow.go
···
}
return &GraphFollow{
+
ID: string(uri),
CreatedAt: out.CreatedAt,
Subject: out.Subject,
}
+1
models/graph_list.go
···
}
list := GraphList{
+
ID: string(uri),
CreatedAt: out.CreatedAt,
Description: out.Description,
Name: out.Name,
+1 -1
cmd/monarch/main.go
···
},
&cli.IntFlag{
Name: "sync-requests-limit",
-
Value: 30,
+
Value: 10, // ratelimit-policy: 3000;w=300
},
}
+4 -20
cmd/monarch/census.go
···
)
type CensusService struct {
-
cursor *CursorService
-
backfill *backfill.Backfiller
-
+
cursor *CursorService
+
backfill *backfill.Backfiller
seenHosts map[string]bool
-
seenLk sync.Mutex
-
-
storeLk sync.Mutex
}
type jobMaker interface {
···
for _, host := range res.Hosts {
// don't reprocess hosts already handled
-
cs.seenLk.Lock()
-
_, ok := cs.seenHosts[host.Hostname]
-
cs.seenLk.Unlock()
-
if ok {
+
seen := cs.seenHosts[host.Hostname]
+
if seen {
slog.Info("already processed host, skipping", "host", host)
continue
}
···
return
}
-
cs.storeLk.Lock()
hcur, err := cs.cursor.GetHostCursor(host)
if err != nil {
slog.Error("error fetching host cursor", "err", err)
}
-
cs.storeLk.Unlock()
var added int
curs := hcur.Cursor
···
continue
}
-
cs.storeLk.Lock()
for _, repo := range res.Repos {
_, err := jmstore.GetOrCreateJob(ctx, repo.Did, backfill.StateEnqueued)
if err != nil {
···
added += 1
}
}
-
cs.storeLk.Unlock()
if res.Cursor != nil && *res.Cursor != "" {
curs = *res.Cursor
-
cs.storeLk.Lock()
if err := cs.cursor.SetHostCursor(host, curs); err != nil {
slog.Error("error updating cursor for host", "err", err)
}
-
cs.storeLk.Unlock()
} else {
break
}
}
slog.Info("finished listing repos", "host", host)
-
-
cs.seenLk.Lock()
-
defer cs.seenLk.Unlock()
-
cs.seenHosts[host] = true
}
+1 -4
cmd/monarch/cursors.go
···
import (
"context"
"log/slog"
-
"sync"
"time"
"gorm.io/gorm"
)
type CursorService struct {
-
store *gorm.DB
-
-
firehoseLk sync.Mutex
+
store *gorm.DB
firehoseSeq int64
}
+3 -1
cmd/monarch/firehose.go
···
func NewFirehoseConnection(ctx context.Context, cctx *cli.Context, cursorSvc *CursorService) (*websocket.Conn, error) {
url := fmt.Sprintf("wss://%s/xrpc/com.atproto.sync.subscribeRepos", cctx.String("relay-host"))
curs, err := cursorSvc.GetFirehoseCursor()
-
if err == nil { // reversed
+
if err != nil {
+
slog.Error("error getting firehose cursor", "err", err)
+
} else if curs > 0 {
url += fmt.Sprintf("?cursor=%d", curs)
}