+4 -4

backfill/resyncer.go

View file Reviewed

··· 51 51 logger: logger.With("component", "resyncer"), 52 52 db: db, 53 53 repos: repos, 54 54 - repoFetchTimeout: 30 * time.Second, 54 54 + repoFetchTimeout: 120 * time.Second, 55 55 collectionFilters: []string{collectionCalendarEvent, collectionCalendarRsvp}, 56 56 parallelism: 2, 57 57 pdsBackoff: make(map[string]time.Time), ··· 247 247 248 248 aturi := syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", did, collStr, rkeyStr)) 249 249 250 250 - // Write the record directly into the project's tables. Errors on a single 251 251 - // record are logged and skipped so one bad record can't block the whole 252 252 - // repo from reaching the active state. 250 250 + // Writes the records to the database. 251 251 + // I do think there's an edge case where if a rsvp happens before the event it is not written to the database. 252 252 + // But just going let that be for now 253 253 switch collStr { 254 254 case collectionCalendarEvent: 255 255 var rec community.CalendarEvent

+2 -1

cmd/main.go

View file Reviewed

··· 50 50 // of leaving the process half-alive. 51 51 var wg sync.WaitGroup 52 52 53 53 - listener := listener.NewListener(db) 53 53 + repoManager := backfill.NewRepoManager(slog.Default(), db, cdir) 54 54 + listener := listener.NewListener(db, repoManager) 54 55 wg.Go(func() { 55 56 defer cancel() 56 57 listener.Run(ctx)

-20

go.mod

View file Reviewed

··· 20 20 ) 21 21 22 22 require ( 23 23 - github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect 24 23 github.com/beorn7/perks v1.0.1 // indirect 25 24 github.com/cespare/xxhash/v2 v2.2.0 // indirect 26 25 github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect 27 27 - github.com/felixge/httpsnoop v1.0.4 // indirect 28 26 github.com/fsnotify/fsnotify v1.9.0 // indirect 29 27 github.com/go-logr/logr v1.4.1 // indirect 30 28 github.com/go-logr/stdr v1.2.2 // indirect 31 29 github.com/go-viper/mapstructure/v2 v2.4.0 // indirect 32 30 github.com/gogo/protobuf v1.3.2 // indirect 33 33 - github.com/golang-jwt/jwt v3.2.2+incompatible // indirect 34 34 - github.com/golang-jwt/jwt/v5 v5.2.2 // indirect 35 31 github.com/google/uuid v1.4.0 // indirect 36 36 - github.com/hashicorp/go-cleanhttp v0.5.2 // indirect 37 37 - github.com/hashicorp/go-retryablehttp v0.7.5 // indirect 38 32 github.com/hashicorp/golang-lru v1.0.2 // indirect 39 33 github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect 40 34 github.com/ipfs/bbloom v0.0.4 // indirect ··· 56 50 github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 // indirect 57 51 github.com/ipld/go-codec-dagpb v1.6.0 // indirect 58 52 github.com/ipld/go-ipld-prime v0.21.0 // indirect 59 59 - github.com/jackc/pgpassfile v1.0.0 // indirect 60 60 - github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect 61 61 - github.com/jackc/pgx/v5 v5.5.0 // indirect 62 62 - github.com/jackc/puddle/v2 v2.2.1 // indirect 63 53 github.com/jbenet/goprocess v0.1.4 // indirect 64 54 github.com/jinzhu/inflection v1.0.0 // indirect 65 55 github.com/jinzhu/now v1.1.5 // indirect 66 56 github.com/klauspost/cpuid/v2 v2.2.7 // indirect 67 67 - github.com/labstack/echo/v4 v4.11.3 // indirect 68 68 - github.com/labstack/gommon v0.4.1 // indirect 69 69 - github.com/mattn/go-colorable v0.1.13 // indirect 70 57 github.com/mattn/go-isatty v0.0.20 // indirect 71 58 github.com/mattn/go-sqlite3 v1.14.44 // indirect 72 59 github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect ··· 84 71 github.com/prometheus/client_model v0.5.0 // indirect 85 72 github.com/prometheus/common v0.45.0 // indirect 86 73 github.com/prometheus/procfs v0.12.0 // indirect 87 87 - github.com/puzpuzpuz/xsync/v4 v4.2.0 // indirect 88 74 github.com/russross/blackfriday/v2 v2.1.0 // indirect 89 75 github.com/sagikazarmark/locafero v0.11.0 // indirect 90 76 github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect ··· 93 79 github.com/spf13/cast v1.10.0 // indirect 94 80 github.com/spf13/pflag v1.0.10 // indirect 95 81 github.com/subosito/gotenv v1.6.0 // indirect 96 96 - github.com/urfave/cli/v3 v3.4.1 // indirect 97 97 - github.com/valyala/bytebufferpool v1.0.0 // indirect 98 98 - github.com/valyala/fasttemplate v1.2.2 // indirect 99 82 github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect 100 83 gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b // indirect 101 84 gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 // indirect 102 102 - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect 103 85 go.opentelemetry.io/otel v1.21.0 // indirect 104 86 go.opentelemetry.io/otel/metric v1.21.0 // indirect 105 87 go.opentelemetry.io/otel/trace v1.21.0 // indirect ··· 109 91 go.yaml.in/yaml/v3 v3.0.4 // indirect 110 92 golang.org/x/crypto v0.50.0 // indirect 111 93 golang.org/x/mod v0.35.0 // indirect 112 112 - golang.org/x/net v0.53.0 // indirect 113 94 golang.org/x/sync v0.20.0 // indirect 114 95 golang.org/x/sys v0.43.0 // indirect 115 96 golang.org/x/text v0.37.0 // indirect 116 97 golang.org/x/time v0.14.0 // indirect 117 98 golang.org/x/tools v0.44.0 // indirect 118 99 google.golang.org/protobuf v1.33.0 // indirect 119 119 - gorm.io/driver/postgres v1.5.7 // indirect 120 100 lukechampine.com/blake3 v1.2.1 // indirect 121 101 )

-45

go.sum

View file Reviewed

··· 1 1 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= 2 2 - github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b h1:5/++qT1/z812ZqBvqQt6ToRswSuPZ/B33m6xVHRzADU= 3 3 - github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b/go.mod h1:4+EPqMRApwwE/6yo6CxiHoSnBzjRr3jsqer7frxP8y4= 4 2 github.com/arran4/golang-ical v0.3.5 h1:bbz6ld4dC+MmCKiFfOd6SkmIGnhNMBACZ485ULh7p9A= 5 3 github.com/arran4/golang-ical v0.3.5/go.mod h1:OnguFgjN0Hmx8jzpmWcC+AkHio94ujmLHKoaef7xQh8= 6 4 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= ··· 24 22 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0/go.mod h1:v57UDF4pDQJcEfFUCRop3lJL149eHGSe9Jvczhzjo/0= 25 23 github.com/earthboundkid/versioninfo/v2 v2.24.1 h1:SJTMHaoUx3GzjjnUO1QzP3ZXK6Ee/nbWyCm58eY3oUg= 26 24 github.com/earthboundkid/versioninfo/v2 v2.24.1/go.mod h1:VcWEooDEuyUJnMfbdTh0uFN4cfEIg+kHMuWB2CDCLjw= 27 27 - github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= 28 28 - github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= 29 25 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= 30 26 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= 31 27 github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= ··· 40 36 github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0= 41 37 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= 42 38 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= 43 43 - github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY= 44 44 - github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I= 45 45 - github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8= 46 46 - github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= 47 39 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= 48 40 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= 49 41 github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8= ··· 55 47 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= 56 48 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= 57 49 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= 58 58 - github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= 59 59 - github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48= 60 60 - github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= 61 61 - github.com/hashicorp/go-retryablehttp v0.7.5 h1:bJj+Pj19UZMIweq/iie+1u5YCdGrnxCT9yvm0e+Nd5M= 62 62 - github.com/hashicorp/go-retryablehttp v0.7.5/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8= 63 50 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= 64 51 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= 65 52 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= ··· 123 110 github.com/ipld/go-codec-dagpb v1.6.0/go.mod h1:ANzFhfP2uMJxRBr8CE+WQWs5UsNa0pYtmKZ+agnUw9s= 124 111 github.com/ipld/go-ipld-prime v0.21.0 h1:n4JmcpOlPDIxBcY037SVfpd1G+Sj1nKZah0m6QH9C2E= 125 112 github.com/ipld/go-ipld-prime v0.21.0/go.mod h1:3RLqy//ERg/y5oShXXdx5YIp50cFGOanyMctpPjsvxQ= 126 126 - github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= 127 127 - github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= 128 128 - github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= 129 129 - github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= 130 130 - github.com/jackc/pgx/v5 v5.5.0 h1:NxstgwndsTRy7eq9/kqYc/BZh5w2hHJV86wjvO+1xPw= 131 131 - github.com/jackc/pgx/v5 v5.5.0/go.mod h1:Ig06C2Vu0t5qXC60W8sqIthScaEnFvojjj9dSljmHRA= 132 132 - github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= 133 133 - github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= 134 113 github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus= 135 114 github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc= 136 115 github.com/jbenet/go-cienv v0.1.0/go.mod h1:TqNnHUmJgXau0nCzC7kXWeotg3J9W34CUv5Djy1+FlA= ··· 159 138 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= 160 139 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= 161 140 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= 162 162 - github.com/labstack/echo/v4 v4.11.3 h1:Upyu3olaqSHkCjs1EJJwQ3WId8b8b1hxbogyommKktM= 163 163 - github.com/labstack/echo/v4 v4.11.3/go.mod h1:UcGuQ8V6ZNRmSweBIJkPvGfwCMIlFmiqrPqiEBfPYws= 164 141 github.com/labstack/echo/v5 v5.1.1 h1:4QkvKoS8ps5ch49t8b72QS9Z581ytgxhTzxuB/CBA2I= 165 142 github.com/labstack/echo/v5 v5.1.1/go.mod h1:SyvlSdObGjRXeQfCCXW/sybkZdOOQZBmpKF0bvALaeo= 166 166 - github.com/labstack/gommon v0.4.1 h1:gqEff0p/hTENGMABzezPoPSRtIh1Cvw0ueMOe0/dfOk= 167 167 - github.com/labstack/gommon v0.4.1/go.mod h1:TyTrpPqxR5KMk8LKVtLmfMjeQ5FEkBYdxLYPw/WfrOM= 168 143 github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8= 169 144 github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg= 170 145 github.com/libp2p/go-cidranger v1.1.0 h1:ewPN8EZ0dd1LSnrtuwd4709PXVcITVeuwbag38yPW7c= ··· 185 160 github.com/libp2p/go-netroute v0.2.0/go.mod h1:Vio7LTzZ+6hoT4CMZi5/6CpY3Snzh2vgZhWgxMNwlQI= 186 161 github.com/libp2p/go-openssl v0.1.0 h1:LBkKEcUv6vtZIQLVTegAil8jbNpJErQ9AnT+bWV+Ooo= 187 162 github.com/libp2p/go-openssl v0.1.0/go.mod h1:OiOxwPpL3n4xlenjx2h7AwSGaFSC/KZvf6gNdOBQMtc= 188 188 - github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= 189 189 - github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= 190 163 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94= 191 191 - github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= 192 164 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= 193 165 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= 194 166 github.com/mattn/go-pointer v0.0.1 h1:n+XhsuGeVO6MEAp7xyEukFINEa+Quek5psIR/ylA6o0= ··· 240 212 github.com/prometheus/common v0.45.0/go.mod h1:YJmSTw9BoKxJplESWWxlbyttQR4uaEcGyv9MZjVOJsY= 241 213 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo= 242 214 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo= 243 243 - github.com/puzpuzpuz/xsync/v4 v4.2.0 h1:dlxm77dZj2c3rxq0/XNvvUKISAmovoXF4a4qM6Wvkr0= 244 244 - github.com/puzpuzpuz/xsync/v4 v4.2.0/go.mod h1:VJDmTCJMBt8igNxnkQd86r+8KUeN1quSfNKu5bLYFQo= 245 215 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= 246 216 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= 247 217 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= ··· 270 240 github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU= 271 241 github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY= 272 242 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= 273 273 - github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= 274 243 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= 275 244 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= 276 245 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= ··· 278 247 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= 279 248 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= 280 249 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= 281 281 - github.com/urfave/cli v1.22.10 h1:p8Fspmz3iTctJstry1PYS3HVdllxnEzTEsgIgtxTrCk= 282 250 github.com/urfave/cli v1.22.10/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= 283 251 github.com/urfave/cli/v2 v2.27.7 h1:bH59vdhbjLv3LAvIu6gd0usJHgoTTPhCFib8qqOwXYU= 284 252 github.com/urfave/cli/v2 v2.27.7/go.mod h1:CyNAG/xg+iAOg0N4MPGZqVmv2rCoP267496AOXUZjA4= 285 285 - github.com/urfave/cli/v3 v3.4.1 h1:1M9UOCy5bLmGnuu1yn3t3CB4rG79Rtoxuv1sPhnm6qM= 286 286 - github.com/urfave/cli/v3 v3.4.1/go.mod h1:FJSKtM/9AiiTOJL4fJ6TbMUkxBXn7GO9guZqoZtpYpo= 287 287 - github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= 288 288 - github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= 289 289 - github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo= 290 290 - github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ= 291 253 github.com/warpfork/go-testmark v0.12.1 h1:rMgCpJfwy1sJ50x0M0NgyphxYYPMOODIJHhsXyEHU0s= 292 254 github.com/warpfork/go-testmark v0.12.1/go.mod h1:kHwy7wfvGSPh1rQJYKayD4AbtNaeyZdcGi9tNJTaa5Y= 293 255 github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0 h1:GDDkbFiaK8jsSDJfjId/PEGEShv6ugrt4kYsC5UIDaQ= ··· 303 265 gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b/go.mod h1:/y/V339mxv2sZmYYR64O07VuCpdNZqCTwO8ZcouTMI8= 304 266 gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 h1:qwDnMxjkyLmAFgcfgTnfJrmYKWhHnci3GjDqcZp1M3Q= 305 267 gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02/go.mod h1:JTnUj0mpYiAsuZLmKjTx/ex3AtMowcCgnE7YNyCEP0I= 306 306 - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 h1:aFJWCqJMNjENlcleuuOkGAPH82y0yULBScfXcIEdS24= 307 307 - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1/go.mod h1:sEGXWArGqc3tVa+ekntsN65DmVbVeW+7lTKTjZF3/Fo= 308 268 go.opentelemetry.io/otel v1.21.0 h1:hzLeKBZEL7Okw2mGzZ0cc4k/A7Fta0uoPgaJCr8fsFc= 309 269 go.opentelemetry.io/otel v1.21.0/go.mod h1:QZzNPQPm1zLX4gZK4cMi+71eaorMSGT3A4znnUvNNEo= 310 270 go.opentelemetry.io/otel/metric v1.21.0 h1:tlYWfeo+Bocx5kLEloTjbcDwBuELRrIFxwdQ36PlJu4= ··· 333 293 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= 334 294 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= 335 295 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= 336 336 - golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA= 337 337 - golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs= 338 296 golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= 339 297 golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= 340 298 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= ··· 365 323 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= 366 324 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 367 325 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 368 368 - golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 369 326 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 370 327 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 371 328 golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= ··· 410 367 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= 411 368 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= 412 369 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= 413 413 - gorm.io/driver/postgres v1.5.7 h1:8ptbNJTDbEmhdr62uReG5BGkdQyeasu/FZHxI0IMGnM= 414 414 - gorm.io/driver/postgres v1.5.7/go.mod h1:3e019WlBaYI5o5LIdNV+LyxCMNtLOQETBXL2h4chKpA= 415 370 gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ= 416 371 gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8= 417 372 gorm.io/gorm v1.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg=

+114 -10

listener/listener.go

View file Reviewed

··· 4 4 "context" 5 5 "encoding/json" 6 6 "fmt" 7 7 + "log" 7 8 "log/slog" 8 9 "net/url" 9 10 "strconv" ··· 12 13 13 14 "tangled.org/pds.dad/whens-lunch/models" 14 15 16 16 + comatproto "github.com/bluesky-social/indigo/api/atproto" 15 17 "github.com/gorilla/websocket" 16 18 "github.com/klauspost/compress/zstd" 17 19 "github.com/spf13/viper" 18 20 "gorm.io/gorm" 19 21 "gorm.io/gorm/clause" 22 22 + backfill "tangled.org/pds.dad/whens-lunch/backfill" 20 23 ) 21 24 22 25 const ( ··· 26 29 ) 27 30 28 31 type Listener struct { 29 29 - db *gorm.DB 32 32 + db *gorm.DB 33 33 + repoManager *backfill.RepoManager 30 34 } 31 35 32 32 - func NewListener(db *gorm.DB) *Listener { 33 33 - return &Listener{db: db} 36 36 + func NewListener(db *gorm.DB, repoManager *backfill.RepoManager) *Listener { 37 37 + return &Listener{db: db, repoManager: repoManager} 34 38 } 35 39 36 40 // Sets up a websocket to listen to a JetStream instance for new events. ··· 57 61 58 62 backoff := minReconnectDelay 59 63 for ctx.Err() == nil { 60 60 - connected, err := connectAndStream(ctx, l.db, base, host, decoder, &lastCursor) 64 64 + connected, err := l.connectAndStream(ctx, base, host, decoder, &lastCursor) 61 65 // Persist our latest position whenever a connection ends. 62 66 saveCursor(ctx, l.db, host, lastCursor.Load()) 63 67 if ctx.Err() != nil { ··· 81 85 82 86 // Connects to the JetStream instance and streams events until the connection 83 87 // drops or ctx is cancelled. 84 84 - func connectAndStream(ctx context.Context, db *gorm.DB, base *url.URL, host string, decoder *zstd.Decoder, lastCursor *atomic.Int64) (connected bool, err error) { 88 88 + func (l *Listener) connectAndStream(ctx context.Context, base *url.URL, host string, decoder *zstd.Decoder, lastCursor *atomic.Int64) (connected bool, err error) { 85 89 u := *base // copy so we don't mutate the shared base URL 86 90 query := u.Query() 87 91 query.Add("wantedCollections", "community.lexicon.calendar.*") ··· 139 143 140 144 switch event.Kind { 141 145 case EventKindCommit: 142 142 - handleCommit(ctx, db, &event) 146 146 + l.handleCommit(ctx, &event) 143 147 case EventKindAccount: 144 148 // log.Printf("account: %s", event.Did) 145 149 case EventKindIdentity: ··· 160 164 case <-done: 161 165 return true, nil // dialed successfully; the read loop ended 162 166 case <-ticker.C: 163 163 - saveCursor(ctx, db, host, lastCursor.Load()) 167 167 + saveCursor(ctx, l.db, host, lastCursor.Load()) 164 168 } 165 169 } 166 170 } 167 171 168 172 // handleCommit dispatches a commit event to the appropriate handler based on 169 173 // the record's lexicon collection. Collections we don't care about are ignored. 170 170 - func handleCommit(ctx context.Context, db *gorm.DB, event *Event) { 174 174 + func (l *Listener) handleCommit(ctx context.Context, event *Event) { 171 175 if event.Commit == nil { 172 176 return 173 177 } 174 178 switch event.Commit.Collection { 175 179 case CollectionCalendarEvent: 176 176 - handleCalendarEvent(ctx, db, event.Did, event) 180 180 + if err := l.SaveOrUpdateRepo(ctx, event.Did); err != nil { 181 181 + log.Printf("failed to save or update repo: %v", err) 182 182 + } 183 183 + handleCalendarEvent(ctx, l.db, event.Did, event) 177 184 case CollectionCalendarRsvp: 178 178 - handleCalendarRsvp(ctx, db, event.Did, event.Commit) 185 185 + if err := l.SaveOrUpdateRepo(ctx, event.Did); err != nil { 186 186 + log.Printf("failed to save or update repo: %v", err) 187 187 + } 188 188 + handleCalendarRsvp(ctx, l.db, event.Did, event.Commit) 179 189 } 180 190 } 181 191 ··· 212 222 slog.Error("listener: save cursor", "err", err) 213 223 } 214 224 } 225 225 + 226 226 + func (l *Listener) SaveOrUpdateRepo(ctx context.Context, did string) error { 227 227 + curr, err := l.repoManager.GetRepoState(ctx, did) 228 228 + if err != nil { 229 229 + return err 230 230 + } else if curr == nil { 231 231 + if err := l.repoManager.EnsureRepo(ctx, did); err != nil { 232 232 + return err 233 233 + } 234 234 + } 235 235 + return nil 236 236 + } 237 237 + 238 238 + func (l *Listener) ProcessIdentity(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Identity) error { 239 239 + 240 240 + curr, err := l.repoManager.GetRepoState(ctx, evt.Did) 241 241 + if err != nil { 242 242 + return err 243 243 + } else if curr == nil { 244 244 + // We don't want to save and update repo if we have never seen it before 245 245 + return nil 246 246 + } 247 247 + 248 248 + if err := l.repoManager.RefreshIdentity(ctx, evt.Did); err != nil { 249 249 + return err 250 250 + } 251 251 + 252 252 + return nil 253 253 + } 254 254 + 255 255 + // func (fp *FirehoseProcessor) ProcessAccount(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Account) error { 256 256 + // firehoseEventsReceived.Inc() 257 257 + // defer fp.updateLastSeq(evt.Seq) 258 258 + 259 259 + // curr, err := fp.repos.GetRepoState(ctx, evt.Did) 260 260 + // if err != nil { 261 261 + // return err 262 262 + // } else if curr == nil { 263 263 + // if fp.fullNetworkMode && evt.Active { 264 264 + // if err := fp.repos.EnsureRepo(ctx, evt.Did); err != nil { 265 265 + // fp.logger.Error("failed to auto-track repo", "did", evt.Did, "error", err) 266 266 + // return err 267 267 + // } 268 268 + // firehoseEventsSkipped.Inc() 269 269 + // return nil 270 270 + // } 271 271 + // firehoseEventsSkipped.Inc() 272 272 + // return nil 273 273 + // } 274 274 + 275 275 + // var updateTo models.AccountStatus 276 276 + // if evt.Active { 277 277 + // updateTo = models.AccountStatusActive 278 278 + // } else if evt.Status != nil && (*evt.Status == string(models.AccountStatusDeactivated) || *evt.Status == string(models.AccountStatusTakendown) || *evt.Status == string(models.AccountStatusSuspended) || *evt.Status == string(models.AccountStatusDeleted)) { 279 279 + // updateTo = models.AccountStatus(*evt.Status) 280 280 + // } else { 281 281 + // // no-op for other events such as throttled or desynchronized 282 282 + // firehoseEventsSkipped.Inc() 283 283 + // return nil 284 284 + // } 285 285 + 286 286 + // if curr.Status == updateTo { 287 287 + // firehoseEventsSkipped.Inc() 288 288 + // return nil 289 289 + // } 290 290 + 291 291 + // identityEvt := &IdentityEvt{ 292 292 + // Did: curr.Did, 293 293 + // Handle: curr.Handle, 294 294 + // IsActive: evt.Active, 295 295 + // Status: updateTo, 296 296 + // } 297 297 + 298 298 + // if updateTo == models.AccountStatusDeleted { 299 299 + // if err := fp.events.AddIdentityEvent(ctx, identityEvt, func(tx *gorm.DB) error { 300 300 + // return deleteRepo(tx, evt.Did) 301 301 + // }); err != nil { 302 302 + // fp.logger.Error("failed to delete repo", "did", evt.Did, "error", err) 303 303 + // return err 304 304 + // } 305 305 + // } else { 306 306 + // if err := fp.events.AddIdentityEvent(ctx, identityEvt, func(tx *gorm.DB) error { 307 307 + // return tx.Model(&models.Repo{}). 308 308 + // Where("did = ?", evt.Did). 309 309 + // Update("status", updateTo).Error 310 310 + // }); err != nil { 311 311 + // fp.logger.Error("failed to update repo status", "did", evt.Did, "status", updateTo, "error", err) 312 312 + // return err 313 313 + // } 314 314 + // } 315 315 + 316 316 + // firehoseEventsProcessed.Inc() 317 317 + // return nil 318 318 + // }

-37

models/models.go

View file Reviewed

··· 113 113 return nil 114 114 115 115 } 116 116 - 117 117 - // type RepoState string 118 118 - 119 119 - // const ( 120 120 - // RepoStatePending RepoState = "pending" 121 121 - // RepoStateDesynchronized RepoState = "desynchronized" 122 122 - // RepoStateResyncing RepoState = "resyncing" 123 123 - // // RepoStateActive RepoState = "active" 124 124 - // RepoStateTakendown RepoState = "takendown" 125 125 - // RepoStateSuspended RepoState = "suspended" 126 126 - // RepoStateDeactivated RepoState = "deactivated" 127 127 - // RepoStateError RepoState = "error" 128 128 - // //New state since backfill will be a one time deal 129 129 - // RepoStateDone RepoState = "done" 130 130 - // ) 131 131 - 132 132 - // type AccountStatus string 133 133 - 134 134 - // const ( 135 135 - // AccountStatusActive AccountStatus = "active" 136 136 - // AccountStatusTakendown AccountStatus = "takendown" 137 137 - // AccountStatusSuspended AccountStatus = "suspended" 138 138 - // AccountStatusDeactivated AccountStatus = "deactivated" 139 139 - // AccountStatusDeleted AccountStatus = "deleted" 140 140 - // ) 141 141 - 142 142 - // type Repo struct { 143 143 - // Did string `gorm:"primaryKey"` 144 144 - // State RepoState `gorm:"not null;default:'pending';index:idx_repos_state_retry"` 145 145 - // Status AccountStatus `gorm:"not null;default:'active'"` 146 146 - // Handle string `gorm:"type:text"` 147 147 - // Rev string `gorm:"type:text"` 148 148 - // PrevData string `gorm:"type:text"` 149 149 - // ErrorMsg string `gorm:"type:text"` 150 150 - // RetryCount int `gorm:"not null;default:0"` 151 151 - // RetryAfter int64 `gorm:"not null;default:0;index:idx_repos_state_retry"` // Unix timestamp (seconds) 152 152 - // }