+35

cmd/recordcollector/main.go

··· 23 23 "github.com/edavis/recordcollector/pkg" 24 24 _ "github.com/joho/godotenv/autoload" 25 25 _ "github.com/mattn/go-sqlite3" 26 26 + "github.com/redis/go-redis/v9" 26 27 ) 27 28 28 29 var RecordCollectorDid = os.Getenv("DID") ··· 50 51 logger *slog.Logger 51 52 client *xrpc.Client // client that makes Ozone requests 52 53 db *sql.DB // sqlite3 DB for tracking labels (to avoid dupes) 54 54 + redis *redis.Client 53 55 lk sync.Mutex 54 56 count int 55 57 } ··· 95 97 log.Fatalf("could not init database: %v", err) 96 98 } 97 99 100 100 + redisClient := redis.NewClient(&redis.Options{ 101 101 + Addr: os.Getenv("REDIS_ADDR"), 102 102 + }) 103 103 + if err := redisClient.Ping(ctx).Err(); err != nil { 104 104 + log.Fatalf("could not connect to redis: %v", err) 105 105 + } 106 106 + defer func() { 107 107 + if err := redisClient.Close(); err != nil { 108 108 + logger.Error("failed to close redis", "err", err) 109 109 + } 110 110 + logger.Info("redis closed") 111 111 + }() 112 112 + 98 113 h := &handler{ 99 114 logger: logger, 100 115 client: &ozoneClient, 101 116 db: dbCnx, 117 117 + redis: redisClient, 102 118 } 103 119 h.runRefreshSession(ctx) 104 120 ··· 175 191 } 176 192 177 193 func (h *handler) addLabel(ctx context.Context, did, label string) error { 194 194 + key := fmt.Sprintf("recordcollector_%s_%s", did, label) 195 195 + 196 196 + // Check if already labeled recently 197 197 + exists, err := h.redis.Exists(ctx, key).Result() 198 198 + if err != nil { 199 199 + h.logger.Error("redis error checking key", "err", err, "key", key) 200 200 + // Continue anyway - better to potentially duplicate than miss labels 201 201 + } 202 202 + if exists > 0 { 203 203 + h.logger.Info("skipping label, already emitted recently", "did", did, "label", label) 204 204 + return nil 205 205 + } 206 206 + 178 207 var labelDuration int64 = 24 * 30 179 208 input := &toolsozone.ModerationEmitEvent_Input{ 180 209 CreatedBy: RecordCollectorDid, ··· 198 227 return fmt.Errorf("failed to add label: %w", err) 199 228 } 200 229 h.logger.Info("added label", "view", view) 230 230 + 231 231 + // Set key with 24h TTL after successful emission 232 232 + if err := h.redis.Set(ctx, key, "1", 24*time.Hour).Err(); err != nil { 233 233 + h.logger.Error("failed to set redis key", "err", err, "key", key) 234 234 + // Don't return error - label was emitted successfully 235 235 + } 201 236 202 237 return nil 203 238 }

+11

compose.yaml

··· 10 10 - type: bind 11 11 source: ./labels.db 12 12 target: /labels.db 13 13 + depends_on: 14 14 + - redis 15 15 + 16 16 + redis: 17 17 + image: redis:7-alpine 18 18 + container_name: recordcollector-redis 19 19 + volumes: 20 20 + - redis-data:/data 21 21 + 22 22 + volumes: 23 23 + redis-data:

+2

go.mod

··· 9 9 github.com/bluesky-social/jetstream v0.0.0-20241210005130-ea96859b93d1 10 10 github.com/joho/godotenv v1.5.1 11 11 github.com/mattn/go-sqlite3 v1.14.22 12 12 + github.com/redis/go-redis/v9 v9.17.3 12 13 golang.org/x/text v0.33.0 13 14 ) 14 15 ··· 16 17 github.com/beorn7/perks v1.0.1 // indirect 17 18 github.com/carlmjohnson/versioninfo v0.22.5 // indirect 18 19 github.com/cespare/xxhash/v2 v2.3.0 // indirect 20 20 + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect 19 21 github.com/felixge/httpsnoop v1.0.4 // indirect 20 22 github.com/go-logr/logr v1.4.1 // indirect 21 23 github.com/go-logr/stdr v1.2.2 // indirect

+8

go.sum

··· 6 6 github.com/bluesky-social/indigo v0.0.0-20250320052052-4873aceeabf4/go.mod h1:NVBwZvbBSa93kfyweAmKwOLYawdVHdwZ9s+GZtBBVLA= 7 7 github.com/bluesky-social/jetstream v0.0.0-20241210005130-ea96859b93d1 h1:CFvRtYNSnWRAi/98M3O466t9dYuwtesNbu6FVPymRrA= 8 8 github.com/bluesky-social/jetstream v0.0.0-20241210005130-ea96859b93d1/go.mod h1:WiYEeyJSdUwqoaZ71KJSpTblemUCpwJfh5oVXplK6T4= 9 9 + github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= 10 10 + github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= 11 11 + github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= 12 12 + github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= 9 13 github.com/carlmjohnson/versioninfo v0.22.5 h1:O00sjOLUAFxYQjlN/bzYTuZiS0y6fWDQjMRvwtKgwwc= 10 14 github.com/carlmjohnson/versioninfo v0.22.5/go.mod h1:QT9mph3wcVfISUKd0i9sZfVrPviHuSF+cUtLjm2WSf8= 11 15 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= ··· 14 18 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= 15 19 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= 16 20 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= 21 21 + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= 22 22 + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= 17 23 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= 18 24 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= 19 25 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= ··· 124 130 github.com/prometheus/common v0.54.0/go.mod h1:/TQgMJP5CuVYveyT7n/0Ix8yLNNXy9yRSkhnLTHPDIQ= 125 131 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc= 126 132 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= 133 133 + github.com/redis/go-redis/v9 v9.17.3 h1:fN29NdNrE17KttK5Ndf20buqfDZwGNgoUr9qjl1DQx4= 134 134 + github.com/redis/go-redis/v9 v9.17.3/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370= 127 135 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= 128 136 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= 129 137 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=