aykhans/bsky-feedgen
1
0
Fork 0
mirror of https://github.com/aykhans/bsky-feedgen.git synced 2025-06-07 13:09:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: aykhans:c4bd6affa6a2d06b8c9a34ff3a3d5d9ebdeb61d1
Branches Tags
aykhans:main
aykhans:feat/add-versions
aykhans:feat/add-auth-middleware
..
compare: aykhans:588cfc0fccbb35579f81cf676fb6d791817d9013
Branches Tags
aykhans:main
aykhans:feat/add-versions
aykhans:feat/add-auth-middleware

No commits in common. "c4bd6affa6a2d06b8c9a34ff3a3d5d9ebdeb61d1" and "588cfc0fccbb35579f81cf676fb6d791817d9013" have entirely different histories.
c4bd6affa6 ... 588cfc0fcc

2 changed files with 3 additions and 20 deletions
Show Stats Expand all files Collapse all files

14
pkg/consumer/base.go
View File

@ -173,12 +173,11 @@ func ConsumeAndSaveToMongoDB(
sequenceCursor = nil sequenceCursor = nil
} }
consumerLastFlushingTime := time.Now()
go func() { go func() {
defer cancel() defer cancel()
for { for {
err := RunFirehoseConsumer( err := RunFirehoseConsumer(
localCtx, ctx,
relayHost, relayHost,
func(sequence int64, did syntax.DID, recordKey syntax.RecordKey, post bsky.FeedPost) { func(sequence int64, did syntax.DID, recordKey syntax.RecordKey, post bsky.FeedPost) {
firehoseDataChan <- CallbackData{sequence, did, recordKey, post} firehoseDataChan <- CallbackData{sequence, did, recordKey, post}
@ -187,7 +186,7 @@ func ConsumeAndSaveToMongoDB(
) )
if err != nil { if err != nil {
if localCtx.Err() != nil { if ctx.Err() != nil {
break break
} }
logger.Log.Error(err.Error()) logger.Log.Error(err.Error())
@ -215,7 +214,6 @@ func ConsumeAndSaveToMongoDB(
return nil return nil
case <-localCtx.Done(): case <-localCtx.Done():
logger.Log.Error("inactive firehose consumer error")
return nil return nil
case data := <-firehoseDataChan: case data := <-firehoseDataChan:
@ -263,18 +261,12 @@ func ConsumeAndSaveToMongoDB(
case <-ticker.C: case <-ticker.C:
if len(postBatch) > 0 { if len(postBatch) > 0 {
consumerLastFlushingTime = time.Now() // logger.Log.Info("flushing post batch", "count", len(postBatch))
logger.Log.Info("flushing post batch", "count", len(postBatch))
err := postCollection.Insert(ctx, true, postBatch...) err := postCollection.Insert(ctx, true, postBatch...)
if err != nil { if err != nil {
return fmt.Errorf("mongodb post insert error: %v", err) return fmt.Errorf("mongodb post insert error: %v", err)
} }
postBatch = []*collections.Post{} // Clear batch after insert postBatch = []*collections.Post{} // Clear batch after insert
} else {
// If we haven't seen any data for 25 seconds, cancel the consumer connection
if consumerLastFlushingTime.Add(time.Second*25).Before(time.Now()) {
cancel()
}
} }
} }
} }

9
pkg/generator/az.go
View File

@ -13,11 +13,6 @@ import (
"go.mongodb.org/mongo-driver/mongo/options" "go.mongodb.org/mongo-driver/mongo/options"
) )
var azInvalidUser []string = []string{
"did:plc:5zww7zorx2ajw7hqrhuix3ba",
"did:plc:c4vhz47h566t2ntgd7gtawen",
}
var azValidUsers []string = []string{ var azValidUsers []string = []string{
"did:plc:jbt4qi6psd7rutwzedtecsq7", "did:plc:jbt4qi6psd7rutwzedtecsq7",
"did:plc:yzgdpxsklrmfgqmjghdvw3ti", "did:plc:yzgdpxsklrmfgqmjghdvw3ti",
@ -129,10 +124,6 @@ func (generator *FeedGeneratorAz) IsValid(post *collections.Post) bool {
return false return false
} }
if slices.Contains(azInvalidUser, post.DID) {
return false
}
if slices.Contains(azValidUsers, post.DID) || // Posts from always-valid users if slices.Contains(azValidUsers, post.DID) || // Posts from always-valid users
(slices.Contains(post.Langs, "az") && len(post.Langs) < 3) || // Posts in Azerbaijani language with fewer than 3 languages (slices.Contains(post.Langs, "az") && len(post.Langs) < 3) || // Posts in Azerbaijani language with fewer than 3 languages
generator.textRegex.MatchString(post.Text) { // Posts containing Azerbaijan-related keywords generator.textRegex.MatchString(post.Text) { // Posts containing Azerbaijan-related keywords