commit 56e4a6caa9e0c4d17295f4aedf83f8db4f499ffa · edavis.dev/monarch

+4 -3

cmd/monarch/census.go

···

       161
       161
        
       		}

     

       162
       162
        
       	}

     

       163
       163
        
       

     

       164
       164
       +
       	slog.Info("finished listing repos", "host", host)

     

       165
       165
       +
       

     

       164
       166
        
       	cs.seenLk.Lock()

     

       165
       167
        
       	defer cs.seenLk.Unlock()

     

       168
       168
       +
       

     

       166
       169
        
       	cs.seenHosts[host] = true

     

       167
       167
       -
       

     

       168
       168
       -
       	slog.Info("finished listing repos", "host", host)

     

       169
       170
        
       }

     

       170
       171
        
       

     

       171
       172
        
       func (cs *CensusService) Start(ctx context.Context, cctx *cli.Context) {

     
···

       173
       174
        
       	cs.listHosts(ctx, cctx)

     

       174
       175
        
       

     

       175
       176
        
       	slog.Info("finished with initial refresh, starting ticker")

     

       176
       176
       -
       	t := time.NewTicker(time.Hour)

     

       177
       177
       +
       	t := time.NewTicker(time.Minute * 5)

     

       177
       178
        
       	defer t.Stop()

     

       178
       179
        
       

     

       179
       180
        
       	for {

+1 -6

cmd/monarch/cursors.go

···

       33
       33
        
       		select {

     

       34
       34
        
       		case <-ctx.Done():

     

       35
       35
        
       			slog.Info("stopping cursor checkpointer", "err", ctx.Err())

     

       36
       36
       -
       

     

       37
       37
       -
       			slog.Info("persisting firehose cursor before exit", "seq", cs.firehoseSeq)

     

       38
       38
       -
       			if err := cs.PersistFirehoseCursor(); err != nil {

     

       39
       39
       -
       				slog.Error("error persisting firehose cursor", "err", err)

     

       40
       40
       -
       			}

     

       41
       41
       -
       

     

       36
       36
       +
       			cs.PersistFirehoseCursor() // one final save

     

       42
       37
        
       			return

     

       43
       38
        
       		case <-t.C:

     

       44
       39
        
       		}

+18 -41

cmd/monarch/handlers.go

···

       39
       39
        
       	}

     

       40
       40
        
       }

     

       41
       41
        
       

     

       42
       42
       -
       var validCollections = map[syntax.NSID]bool{

     

       43
       43
       -
       	syntax.NSID("app.bsky.actor.profile"):   false,

     

       44
       44
       -
       	syntax.NSID("app.bsky.feed.generator"):  true,

     

       45
       45
       -
       	syntax.NSID("app.bsky.labeler.service"): true,

     

       46
       46
       -
       }

     

       47
       47
       -
       

     

       48
       48
       -
       func jsonEncode(model any, data *[]byte) ([]byte, error) {

     

       49
       49
       -
       	var (

     

       50
       50
       -
       		body []byte

     

       51
       51
       -
       		err  error

     

       52
       52
       -
       	)

     

       53
       53
       -
       

     

       54
       54
       -
       	switch model.(type) {

     

       55
       55
       -
       	case appbsky.ActorProfile:

     

       56
       56
       -
       		var out appbsky.ActorProfile

     

       57
       57
       -
       		out.UnmarshalCBOR(bytes.NewReader(*data))

     

       58
       58
       -
       		body, err = json.Marshal(out)

     

       59
       59
       -
       

     

       60
       60
       -
       	case appbsky.FeedGenerator:

     

       61
       61
       -
       		var out appbsky.FeedGenerator

     

       62
       62
       -
       		out.UnmarshalCBOR(bytes.NewReader(*data))

     

       63
       63
       -
       		body, err = json.Marshal(out)

     

       64
       64
       -
       

     

       65
       65
       -
       	case appbsky.LabelerService:

     

       66
       66
       -
       		var out appbsky.LabelerService

     

       67
       67
       -
       		out.UnmarshalCBOR(bytes.NewReader(*data))

     

       68
       68
       -
       		body, err = json.Marshal(out)

     

       69
       69
       -
       	}

     

       70
       70
       -
       

     

       71
       71
       -
       	return body, err

     

       42
       42
       +
       var trackedCollections = map[syntax.NSID]bool{

     

       43
       43
       +
       	syntax.NSID("app.bsky.actor.profile"):      false,

     

       44
       44
       +
       	syntax.NSID("app.bsky.feed.generator"):     true,

     

       45
       45
       +
       	syntax.NSID("app.bsky.labeler.service"):    true,

     

       46
       46
       +
       	syntax.NSID("app.bsky.graph.list"):         true,

     

       47
       47
       +
       	syntax.NSID("app.bsky.graph.verification"): true,

     

       48
       48
       +
       	syntax.NSID("app.bsky.graph.starterpack"):  true,

     

       72
       49
        
       }

     

       73
       50
        
       

     

       74
       51
        
       func (hs *HandlerService) HandleUpsert(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid, action Action) error {

     
···

       77
       54
        
       		return fmt.Errorf("error parsing at-uri: %w", err)

     

       78
       55
        
       	}

     

       79
       56
        
       

     

       80
       80
       -
       	valid := validCollections[uri.Collection()]

     

       81
       81
       -
       	if !valid {

     

       57
       57
       +
       	tracked := trackedCollections[uri.Collection()]

     

       58
       58
       +
       	if !tracked {

     

       82
       59
        
       		return nil

     

       83
       60
        
       	}

     

       84
       61
        
       

     

       85
       62
        
       	var body []byte

     

       86
       63
        
       	switch uri.Collection() {

     

       87
       64
        
       	case syntax.NSID("app.bsky.actor.profile"):

     

       88
       88
       -
       		if body, err = jsonEncode(appbsky.ActorProfile{}, rec); err != nil {

     

       89
       89
       -
       			return fmt.Errorf("error parsing actor profile: %w", err)

     

       90
       90
       -
       		}

     

       65
       65
       +
       		var out appbsky.ActorProfile

     

       66
       66
       +
       		out.UnmarshalCBOR(bytes.NewReader(*rec))

     

       67
       67
       +
       		body, err = json.Marshal(out)

     

       91
       68
        
       

     

       92
       69
        
       	case syntax.NSID("app.bsky.feed.generator"):

     

       93
       93
       -
       		if body, err = jsonEncode(appbsky.FeedGenerator{}, rec); err != nil {

     

       94
       94
       -
       			return fmt.Errorf("error parsing feed generator: %w", err)

     

       95
       95
       -
       		}

     

       70
       70
       +
       		var out appbsky.FeedGenerator

     

       71
       71
       +
       		out.UnmarshalCBOR(bytes.NewReader(*rec))

     

       72
       72
       +
       		body, err = json.Marshal(out)

     

       96
       73
        
       

     

       97
       74
        
       	case syntax.NSID("app.bsky.labeler.service"):

     

       98
       98
       -
       		if body, err = jsonEncode(appbsky.LabelerService{}, rec); err != nil {

     

       99
       99
       -
       			return fmt.Errorf("error parsing labeler service: %w", err)

     

       100
       100
       -
       		}

     

       75
       75
       +
       		var out appbsky.LabelerService

     

       76
       76
       +
       		out.UnmarshalCBOR(bytes.NewReader(*rec))

     

       77
       77
       +
       		body, err = json.Marshal(out)

     

       101
       78
        
       	}

     

       102
       79
        
       

     

       103
       80
        
       	switch action {