diff --git a/cmd/cocoon/main.go b/cmd/cocoon/main.go index c1b7630..acc96ea 100644 --- a/cmd/cocoon/main.go +++ b/cmd/cocoon/main.go @@ -9,6 +9,7 @@ import ( "os" "time" + "github.com/bluesky-social/go-util/pkg/telemetry" "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/haileyok/cocoon/internal/helpers" @@ -154,6 +155,8 @@ func main() { Name: "fallback-proxy", EnvVars: []string{"COCOON_FALLBACK_PROXY"}, }, + telemetry.CLIFlagDebug, + telemetry.CLIFlagMetricsListenAddress, }, Commands: []*cli.Command{ runServe, @@ -177,7 +180,11 @@ var runServe = &cli.Command{ Flags: []cli.Flag{}, Action: func(cmd *cli.Context) error { + logger := telemetry.StartLogger(cmd) + telemetry.StartMetrics(cmd) + s, err := server.New(&server.Args{ + Logger: logger, Addr: cmd.String("addr"), DbName: cmd.String("db-name"), DbType: cmd.String("db-type"), diff --git a/metrics/metrics.go b/metrics/metrics.go new file mode 100644 index 0000000..4262c2f --- /dev/null +++ b/metrics/metrics.go @@ -0,0 +1,30 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +const ( + NAMESPACE = "cocoon" +) + +var ( + RelaysConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: NAMESPACE, + Name: "relays_connected", + Help: "number of connected relays, by host", + }, []string{"host"}) + + RelaySends = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: NAMESPACE, + Name: "relay_sends", + Help: "number of events sent to a relay, by host", + }, []string{"host"}) + + RepoOperations = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: NAMESPACE, + Name: "repo_operations", + Help: "number of operations made against repos", + }, []string{"kind"}) +) diff --git a/oauth/dpop/jti_cache.go b/oauth/dpop/jti_cache.go index 212c3d3..cc6c7af 100644 --- a/oauth/dpop/jti_cache.go +++ b/oauth/dpop/jti_cache.go @@ -14,7 +14,7 @@ type jtiCache struct { } func newJTICache(size int) *jtiCache { - cache := cache.NewCache[string, bool]().WithTTL(24 * time.Hour).WithLRU().WithTTL(constants.JTITtl) + cache := cache.NewCache[string, bool]().WithTTL(24 * time.Hour).WithLRU().WithTTL(constants.JTITtl).WithMaxKeys(size) return &jtiCache{ cache: cache, mu: sync.Mutex{}, diff --git a/server/handle_account.go b/server/handle_account.go index 87cfbc0..8ab4450 100644 --- a/server/handle_account.go +++ b/server/handle_account.go @@ -12,6 +12,7 @@ import ( func (s *Server) handleAccount(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleAuth") repo, sess, err := s.getSessionRepoOrErr(e) if err != nil { @@ -22,7 +23,7 @@ func (s *Server) handleAccount(e echo.Context) error { var tokens []provider.OauthToken if err := s.db.Raw(ctx, "SELECT * FROM oauth_tokens WHERE sub = ? AND created_at < ? ORDER BY created_at ASC", nil, repo.Repo.Did, oldestPossibleSession).Scan(&tokens).Error; err != nil { - s.logger.Error("couldnt fetch oauth sessions for account", "did", repo.Repo.Did, "error", err) + logger.Error("couldnt fetch oauth sessions for account", "did", repo.Repo.Did, "error", err) sess.AddFlash("Unable to fetch sessions. See server logs for more details.", "error") sess.Save(e.Request(), e.Response()) return e.Render(200, "account.html", map[string]any{ diff --git a/server/handle_account_revoke.go b/server/handle_account_revoke.go index febb66a..c392fc1 100644 --- a/server/handle_account_revoke.go +++ b/server/handle_account_revoke.go @@ -11,10 +11,11 @@ type AccountRevokeInput struct { func (s *Server) handleAccountRevoke(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleAcocuntRevoke") var req AccountRevokeInput if err := e.Bind(&req); err != nil { - s.logger.Error("could not bind account revoke request", "error", err) + logger.Error("could not bind account revoke request", "error", err) return helpers.ServerError(e, nil) } @@ -24,7 +25,7 @@ func (s *Server) handleAccountRevoke(e echo.Context) error { } if err := s.db.Exec(ctx, "DELETE FROM oauth_tokens WHERE sub = ? AND token = ?", nil, repo.Repo.Did, req.Token).Error; err != nil { - s.logger.Error("couldnt delete oauth session for account", "did", repo.Repo.Did, "token", req.Token, "error", err) + logger.Error("couldnt delete oauth session for account", "did", repo.Repo.Did, "token", req.Token, "error", err) sess.AddFlash("Unable to revoke session. See server logs for more details.", "error") sess.Save(e.Request(), e.Response()) return e.Redirect(303, "/account") diff --git a/server/handle_account_signin.go b/server/handle_account_signin.go index 57b082a..a024f00 100644 --- a/server/handle_account_signin.go +++ b/server/handle_account_signin.go @@ -63,10 +63,11 @@ func (s *Server) handleAccountSigninGet(e echo.Context) error { func (s *Server) handleAccountSigninPost(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleAccountSigninPost") var req OauthSigninInput if err := e.Bind(&req); err != nil { - s.logger.Error("error binding sign in req", "error", err) + logger.Error("error binding sign in req", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_identity_get_recommended_did_credentials.go b/server/handle_identity_get_recommended_did_credentials.go index f6a6c47..d9c4360 100644 --- a/server/handle_identity_get_recommended_did_credentials.go +++ b/server/handle_identity_get_recommended_did_credentials.go @@ -8,15 +8,17 @@ import ( ) func (s *Server) handleGetRecommendedDidCredentials(e echo.Context) error { + logger := s.logger.With("name", "handleIdentityGetRecommendedDidCredentials") + repo := e.Get("repo").(*models.RepoActor) k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey) if err != nil { - s.logger.Error("error parsing key", "error", err) + logger.Error("error parsing key", "error", err) return helpers.ServerError(e, nil) } creds, err := s.plcClient.CreateDidCredentials(k, "", repo.Actor.Handle) if err != nil { - s.logger.Error("error crating did credentials", "error", err) + logger.Error("error crating did credentials", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_identity_request_plc_operation.go b/server/handle_identity_request_plc_operation.go index 028e805..16b0ba6 100644 --- a/server/handle_identity_request_plc_operation.go +++ b/server/handle_identity_request_plc_operation.go @@ -11,6 +11,7 @@ import ( func (s *Server) handleIdentityRequestPlcOperationSignature(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleIdentityRequestPlcOperationSignature") urepo := e.Get("repo").(*models.RepoActor) @@ -18,12 +19,12 @@ func (s *Server) handleIdentityRequestPlcOperationSignature(e echo.Context) erro eat := time.Now().Add(10 * time.Minute).UTC() if err := s.db.Exec(ctx, "UPDATE repos SET plc_operation_code = ?, plc_operation_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating user", "error", err) + logger.Error("error updating user", "error", err) return helpers.ServerError(e, nil) } if err := s.sendPlcTokenReset(urepo.Email, urepo.Handle, code); err != nil { - s.logger.Error("error sending mail", "error", err) + logger.Error("error sending mail", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_identity_sign_plc_operation.go b/server/handle_identity_sign_plc_operation.go index feb2bbd..169384e 100644 --- a/server/handle_identity_sign_plc_operation.go +++ b/server/handle_identity_sign_plc_operation.go @@ -27,11 +27,13 @@ type ComAtprotoSignPlcOperationResponse struct { } func (s *Server) handleSignPlcOperation(e echo.Context) error { + logger := s.logger.With("name", "handleSignPlcOperation") + repo := e.Get("repo").(*models.RepoActor) var req ComAtprotoSignPlcOperationRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } @@ -54,7 +56,7 @@ func (s *Server) handleSignPlcOperation(e echo.Context) error { ctx := context.WithValue(e.Request().Context(), "skip-cache", true) log, err := identity.FetchDidAuditLog(ctx, nil, repo.Repo.Did) if err != nil { - s.logger.Error("error fetching doc", "error", err) + logger.Error("error fetching doc", "error", err) return helpers.ServerError(e, nil) } @@ -83,17 +85,17 @@ func (s *Server) handleSignPlcOperation(e echo.Context) error { k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey) if err != nil { - s.logger.Error("error parsing signing key", "error", err) + logger.Error("error parsing signing key", "error", err) return helpers.ServerError(e, nil) } if err := s.plcClient.SignOp(k, &op); err != nil { - s.logger.Error("error signing plc operation", "error", err) + logger.Error("error signing plc operation", "error", err) return helpers.ServerError(e, nil) } if err := s.db.Exec(ctx, "UPDATE repos SET plc_operation_code = NULL, plc_operation_code_expires_at = NULL WHERE did = ?", nil, repo.Repo.Did).Error; err != nil { - s.logger.Error("error updating repo", "error", err) + logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_identity_submit_plc_operation.go b/server/handle_identity_submit_plc_operation.go index 1af3f8f..0b23651 100644 --- a/server/handle_identity_submit_plc_operation.go +++ b/server/handle_identity_submit_plc_operation.go @@ -21,11 +21,13 @@ type ComAtprotoSubmitPlcOperationRequest struct { } func (s *Server) handleSubmitPlcOperation(e echo.Context) error { + logger := s.logger.With("name", "handleIdentitySubmitPlcOperation") + repo := e.Get("repo").(*models.RepoActor) var req ComAtprotoSubmitPlcOperationRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } @@ -40,12 +42,12 @@ func (s *Server) handleSubmitPlcOperation(e echo.Context) error { k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey) if err != nil { - s.logger.Error("error parsing key", "error", err) + logger.Error("error parsing key", "error", err) return helpers.ServerError(e, nil) } required, err := s.plcClient.CreateDidCredentials(k, "", repo.Actor.Handle) if err != nil { - s.logger.Error("error crating did credentials", "error", err) + logger.Error("error crating did credentials", "error", err) return helpers.ServerError(e, nil) } @@ -72,7 +74,7 @@ func (s *Server) handleSubmitPlcOperation(e echo.Context) error { } if err := s.passport.BustDoc(context.TODO(), repo.Repo.Did); err != nil { - s.logger.Warn("error busting did doc", "error", err) + logger.Warn("error busting did doc", "error", err) } s.evtman.AddEvent(context.TODO(), &events.XRPCStreamEvent{ diff --git a/server/handle_identity_update_handle.go b/server/handle_identity_update_handle.go index 93a41c1..9e522e2 100644 --- a/server/handle_identity_update_handle.go +++ b/server/handle_identity_update_handle.go @@ -22,11 +22,13 @@ type ComAtprotoIdentityUpdateHandleRequest struct { } func (s *Server) handleIdentityUpdateHandle(e echo.Context) error { + logger := s.logger.With("name", "handleIdentityUpdateHandle") + repo := e.Get("repo").(*models.RepoActor) var req ComAtprotoIdentityUpdateHandleRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } @@ -41,7 +43,7 @@ func (s *Server) handleIdentityUpdateHandle(e echo.Context) error { if strings.HasPrefix(repo.Repo.Did, "did:plc:") { log, err := identity.FetchDidAuditLog(ctx, nil, repo.Repo.Did) if err != nil { - s.logger.Error("error fetching doc", "error", err) + logger.Error("error fetching doc", "error", err) return helpers.ServerError(e, nil) } @@ -68,7 +70,7 @@ func (s *Server) handleIdentityUpdateHandle(e echo.Context) error { k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey) if err != nil { - s.logger.Error("error parsing signing key", "error", err) + logger.Error("error parsing signing key", "error", err) return helpers.ServerError(e, nil) } @@ -82,7 +84,7 @@ func (s *Server) handleIdentityUpdateHandle(e echo.Context) error { } if err := s.passport.BustDoc(context.TODO(), repo.Repo.Did); err != nil { - s.logger.Warn("error busting did doc", "error", err) + logger.Warn("error busting did doc", "error", err) } s.evtman.AddEvent(context.TODO(), &events.XRPCStreamEvent{ @@ -95,7 +97,7 @@ func (s *Server) handleIdentityUpdateHandle(e echo.Context) error { }) if err := s.db.Exec(ctx, "UPDATE actors SET handle = ? WHERE did = ?", nil, req.Handle, repo.Repo.Did).Error; err != nil { - s.logger.Error("error updating handle in db", "error", err) + logger.Error("error updating handle in db", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_import_repo.go b/server/handle_import_repo.go index ea1b0c5..2ff4433 100644 --- a/server/handle_import_repo.go +++ b/server/handle_import_repo.go @@ -19,12 +19,13 @@ import ( func (s *Server) handleRepoImportRepo(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleImportRepo") urepo := e.Get("repo").(*models.RepoActor) b, err := io.ReadAll(e.Request().Body) if err != nil { - s.logger.Error("could not read bytes in import request", "error", err) + logger.Error("could not read bytes in import request", "error", err) return helpers.ServerError(e, nil) } @@ -32,20 +33,20 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { cs, err := car.NewCarReader(bytes.NewReader(b)) if err != nil { - s.logger.Error("could not read car in import request", "error", err) + logger.Error("could not read car in import request", "error", err) return helpers.ServerError(e, nil) } orderedBlocks := []blocks.Block{} currBlock, err := cs.Next() if err != nil { - s.logger.Error("could not get first block from car", "error", err) + logger.Error("could not get first block from car", "error", err) return helpers.ServerError(e, nil) } currBlockCt := 1 for currBlock != nil { - s.logger.Info("someone is importing their repo", "block", currBlockCt) + logger.Info("someone is importing their repo", "block", currBlockCt) orderedBlocks = append(orderedBlocks, currBlock) next, _ := cs.Next() currBlock = next @@ -55,13 +56,13 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { slices.Reverse(orderedBlocks) if err := bs.PutMany(context.TODO(), orderedBlocks); err != nil { - s.logger.Error("could not insert blocks", "error", err) + logger.Error("could not insert blocks", "error", err) return helpers.ServerError(e, nil) } r, err := repo.OpenRepo(context.TODO(), bs, cs.Header.Roots[0]) if err != nil { - s.logger.Error("could not open repo", "error", err) + logger.Error("could not open repo", "error", err) return helpers.ServerError(e, nil) } @@ -76,7 +77,7 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { cidStr := cid.String() b, err := bs.Get(context.TODO(), cid) if err != nil { - s.logger.Error("record bytes don't exist in blockstore", "error", err) + logger.Error("record bytes don't exist in blockstore", "error", err) return helpers.ServerError(e, nil) } @@ -96,7 +97,7 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { return nil }); err != nil { tx.Rollback() - s.logger.Error("record bytes don't exist in blockstore", "error", err) + logger.Error("record bytes don't exist in blockstore", "error", err) return helpers.ServerError(e, nil) } @@ -104,12 +105,12 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { root, rev, err := r.Commit(context.TODO(), urepo.SignFor) if err != nil { - s.logger.Error("error committing", "error", err) + logger.Error("error committing", "error", err) return helpers.ServerError(e, nil) } if err := s.UpdateRepo(context.TODO(), urepo.Repo.Did, root, rev); err != nil { - s.logger.Error("error updating repo after commit", "error", err) + logger.Error("error updating repo after commit", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_oauth_authorize.go b/server/handle_oauth_authorize.go index 19f5dee..ebc180a 100644 --- a/server/handle_oauth_authorize.go +++ b/server/handle_oauth_authorize.go @@ -75,6 +75,7 @@ type OauthAuthorizePostRequest struct { func (s *Server) handleOauthAuthorizePost(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleOauthAuthorizePost") repo, _, err := s.getSessionRepoOrErr(e) if err != nil { @@ -83,7 +84,7 @@ func (s *Server) handleOauthAuthorizePost(e echo.Context) error { var req OauthAuthorizePostRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding authorize post request", "error", err) + logger.Error("error binding authorize post request", "error", err) return helpers.InputError(e, nil) } @@ -118,7 +119,7 @@ func (s *Server) handleOauthAuthorizePost(e echo.Context) error { code := oauth.GenerateCode() if err := s.db.Exec(ctx, "UPDATE oauth_authorization_requests SET sub = ?, code = ?, accepted = ?, ip = ? WHERE request_id = ?", nil, repo.Repo.Did, code, true, e.RealIP(), reqId).Error; err != nil { - s.logger.Error("error updating authorization request", "error", err) + logger.Error("error updating authorization request", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_oauth_par.go b/server/handle_oauth_par.go index 37682cd..c0c64ae 100644 --- a/server/handle_oauth_par.go +++ b/server/handle_oauth_par.go @@ -20,15 +20,16 @@ type OauthParResponse struct { func (s *Server) handleOauthPar(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleOauthPar") var parRequest provider.ParRequest if err := e.Bind(&parRequest); err != nil { - s.logger.Error("error binding for par request", "error", err) + logger.Error("error binding for par request", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(parRequest); err != nil { - s.logger.Error("missing parameters for par request", "error", err) + logger.Error("missing parameters for par request", "error", err) return helpers.InputError(e, nil) } @@ -45,7 +46,7 @@ func (s *Server) handleOauthPar(e echo.Context) error { "error": "use_dpop_nonce", }) } - s.logger.Error("error getting dpop proof", "error", err) + logger.Error("error getting dpop proof", "error", err) return helpers.InputError(e, nil) } @@ -55,7 +56,7 @@ func (s *Server) handleOauthPar(e echo.Context) error { AllowMissingDpopProof: true, }) if err != nil { - s.logger.Error("error authenticating client", "client_id", parRequest.ClientID, "error", err) + logger.Error("error authenticating client", "client_id", parRequest.ClientID, "error", err) return helpers.InputError(e, to.StringPtr(err.Error())) } @@ -66,13 +67,13 @@ func (s *Server) handleOauthPar(e echo.Context) error { } else { if !client.Metadata.DpopBoundAccessTokens { msg := "dpop bound access tokens are not enabled for this client" - s.logger.Error(msg) + logger.Error(msg) return helpers.InputError(e, &msg) } if dpopProof.JKT != *parRequest.DpopJkt { msg := "supplied dpop jkt does not match header dpop jkt" - s.logger.Error(msg) + logger.Error(msg) return helpers.InputError(e, &msg) } } @@ -89,7 +90,7 @@ func (s *Server) handleOauthPar(e echo.Context) error { } if err := s.db.Create(ctx, authRequest, nil).Error; err != nil { - s.logger.Error("error creating auth request in db", "error", err) + logger.Error("error creating auth request in db", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_oauth_token.go b/server/handle_oauth_token.go index 905454b..d3e373e 100644 --- a/server/handle_oauth_token.go +++ b/server/handle_oauth_token.go @@ -39,10 +39,11 @@ type OauthTokenResponse struct { func (s *Server) handleOauthToken(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleOauthToken") var req OauthTokenRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding token request", "error", err) + logger.Error("error binding token request", "error", err) return helpers.ServerError(e, nil) } @@ -58,7 +59,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { "error": "use_dpop_nonce", }) } - s.logger.Error("error getting dpop proof", "error", err) + logger.Error("error getting dpop proof", "error", err) return helpers.InputError(e, nil) } @@ -66,7 +67,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { AllowMissingDpopProof: true, }) if err != nil { - s.logger.Error("error authenticating client", "client_id", req.ClientID, "error", err) + logger.Error("error authenticating client", "client_id", req.ClientID, "error", err) return helpers.InputError(e, to.StringPtr(err.Error())) } @@ -87,7 +88,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { var authReq provider.OauthAuthorizationRequest // get the lil guy and delete him if err := s.db.Raw(ctx, "DELETE FROM oauth_authorization_requests WHERE code = ? RETURNING *", nil, *req.Code).Scan(&authReq).Error; err != nil { - s.logger.Error("error finding authorization request", "error", err) + logger.Error("error finding authorization request", "error", err) return helpers.ServerError(e, nil) } @@ -112,7 +113,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { case "S256": inputChal, err := base64.RawURLEncoding.DecodeString(*authReq.Parameters.CodeChallenge) if err != nil { - s.logger.Error("error decoding code challenge", "error", err) + logger.Error("error decoding code challenge", "error", err) return helpers.ServerError(e, nil) } @@ -173,7 +174,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { RefreshToken: refreshToken, Ip: authReq.Ip, }, nil).Error; err != nil { - s.logger.Error("error creating token in db", "error", err) + logger.Error("error creating token in db", "error", err) return helpers.ServerError(e, nil) } @@ -202,7 +203,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { var oauthToken provider.OauthToken if err := s.db.Raw(ctx, "SELECT * FROM oauth_tokens WHERE refresh_token = ?", nil, req.RefreshToken).Scan(&oauthToken).Error; err != nil { - s.logger.Error("error finding oauth token by refresh token", "error", err, "refresh_token", req.RefreshToken) + logger.Error("error finding oauth token by refresh token", "error", err, "refresh_token", req.RefreshToken) return helpers.ServerError(e, nil) } @@ -260,7 +261,7 @@ func (s *Server) handleOauthToken(e echo.Context) error { } if err := s.db.Exec(ctx, "UPDATE oauth_tokens SET token = ?, refresh_token = ?, expires_at = ?, updated_at = ? WHERE refresh_token = ?", nil, accessString, nextRefreshToken, eat, now, *req.RefreshToken).Error; err != nil { - s.logger.Error("error updating token", "error", err) + logger.Error("error updating token", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_proxy.go b/server/handle_proxy.go index d602b25..5a56ea9 100644 --- a/server/handle_proxy.go +++ b/server/handle_proxy.go @@ -47,7 +47,7 @@ func (s *Server) getAtprotoProxyEndpointFromRequest(e echo.Context) (string, str } func (s *Server) handleProxy(e echo.Context) error { - lgr := s.logger.With("handler", "handleProxy") + logger := s.logger.With("handler", "handleProxy") repo, isAuthed := e.Get("repo").(*models.RepoActor) @@ -58,7 +58,7 @@ func (s *Server) handleProxy(e echo.Context) error { endpoint, svcDid, err := s.getAtprotoProxyEndpointFromRequest(e) if err != nil { - lgr.Error("could not get atproto proxy", "error", err) + logger.Error("could not get atproto proxy", "error", err) return helpers.ServerError(e, nil) } @@ -90,7 +90,7 @@ func (s *Server) handleProxy(e echo.Context) error { } hj, err := json.Marshal(header) if err != nil { - lgr.Error("error marshaling header", "error", err) + logger.Error("error marshaling header", "error", err) return helpers.ServerError(e, nil) } @@ -118,7 +118,7 @@ func (s *Server) handleProxy(e echo.Context) error { } pj, err := json.Marshal(payload) if err != nil { - lgr.Error("error marashaling payload", "error", err) + logger.Error("error marashaling payload", "error", err) return helpers.ServerError(e, nil) } @@ -129,13 +129,13 @@ func (s *Server) handleProxy(e echo.Context) error { sk, err := secp256k1secec.NewPrivateKey(repo.SigningKey) if err != nil { - lgr.Error("can't load private key", "error", err) + logger.Error("can't load private key", "error", err) return err } R, S, _, err := sk.SignRaw(rand.Reader, hash[:]) if err != nil { - lgr.Error("error signing", "error", err) + logger.Error("error signing", "error", err) } rBytes := R.Bytes() diff --git a/server/handle_repo_apply_writes.go b/server/handle_repo_apply_writes.go index db8f760..479e72b 100644 --- a/server/handle_repo_apply_writes.go +++ b/server/handle_repo_apply_writes.go @@ -27,22 +27,23 @@ type ComAtprotoRepoApplyWritesOutput struct { func (s *Server) handleApplyWrites(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleRepoApplyWrites") var req ComAtprotoRepoApplyWritesInput if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.InputError(e, nil) } repo := e.Get("repo").(*models.RepoActor) if repo.Repo.Did != req.Repo { - s.logger.Warn("mismatched repo/auth") + logger.Warn("mismatched repo/auth") return helpers.InputError(e, nil) } @@ -58,7 +59,7 @@ func (s *Server) handleApplyWrites(e echo.Context) error { results, err := s.repoman.applyWrites(ctx, repo.Repo, ops, req.SwapCommit) if err != nil { - s.logger.Error("error applying writes", "error", err) + logger.Error("error applying writes", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_create_record.go b/server/handle_repo_create_record.go index 5c86c50..ed2f5e2 100644 --- a/server/handle_repo_create_record.go +++ b/server/handle_repo_create_record.go @@ -18,22 +18,23 @@ type ComAtprotoRepoCreateRecordInput struct { func (s *Server) handleCreateRecord(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleCreateRecord") repo := e.Get("repo").(*models.RepoActor) var req ComAtprotoRepoCreateRecordInput if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.InputError(e, nil) } if repo.Repo.Did != req.Repo { - s.logger.Warn("mismatched repo/auth") + logger.Warn("mismatched repo/auth") return helpers.InputError(e, nil) } @@ -53,7 +54,7 @@ func (s *Server) handleCreateRecord(e echo.Context) error { }, }, req.SwapCommit) if err != nil { - s.logger.Error("error applying writes", "error", err) + logger.Error("error applying writes", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_delete_record.go b/server/handle_repo_delete_record.go index ecc22fc..3e60562 100644 --- a/server/handle_repo_delete_record.go +++ b/server/handle_repo_delete_record.go @@ -16,22 +16,23 @@ type ComAtprotoRepoDeleteRecordInput struct { func (s *Server) handleDeleteRecord(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleDeleteRecord") repo := e.Get("repo").(*models.RepoActor) var req ComAtprotoRepoDeleteRecordInput if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.InputError(e, nil) } if repo.Repo.Did != req.Repo { - s.logger.Warn("mismatched repo/auth") + logger.Warn("mismatched repo/auth") return helpers.InputError(e, nil) } @@ -44,7 +45,7 @@ func (s *Server) handleDeleteRecord(e echo.Context) error { }, }, req.SwapCommit) if err != nil { - s.logger.Error("error applying writes", "error", err) + logger.Error("error applying writes", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_describe_repo.go b/server/handle_repo_describe_repo.go index 73b0f98..813c60f 100644 --- a/server/handle_repo_describe_repo.go +++ b/server/handle_repo_describe_repo.go @@ -21,6 +21,7 @@ type ComAtprotoRepoDescribeRepoResponse struct { func (s *Server) handleDescribeRepo(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleDescribeRepo") did := e.QueryParam("repo") repo, err := s.getRepoActorByDid(ctx, did) @@ -29,7 +30,7 @@ func (s *Server) handleDescribeRepo(e echo.Context) error { return helpers.InputError(e, to.StringPtr("RepoNotFound")) } - s.logger.Error("error looking up repo", "error", err) + logger.Error("error looking up repo", "error", err) return helpers.ServerError(e, nil) } @@ -37,7 +38,7 @@ func (s *Server) handleDescribeRepo(e echo.Context) error { diddoc, err := s.passport.FetchDoc(e.Request().Context(), repo.Repo.Did) if err != nil { - s.logger.Error("error fetching diddoc", "error", err) + logger.Error("error fetching diddoc", "error", err) return helpers.ServerError(e, nil) } @@ -67,7 +68,7 @@ func (s *Server) handleDescribeRepo(e echo.Context) error { var records []models.Record if err := s.db.Raw(ctx, "SELECT DISTINCT(nsid) FROM records WHERE did = ?", nil, repo.Repo.Did).Scan(&records).Error; err != nil { - s.logger.Error("error getting collections", "error", err) + logger.Error("error getting collections", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_list_missing_blobs.go b/server/handle_repo_list_missing_blobs.go index ff468e7..5efe38d 100644 --- a/server/handle_repo_list_missing_blobs.go +++ b/server/handle_repo_list_missing_blobs.go @@ -23,6 +23,7 @@ type ComAtprotoRepoListMissingBlobsRecordBlob struct { func (s *Server) handleListMissingBlobs(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleListMissingBlos") urepo := e.Get("repo").(*models.RepoActor) @@ -38,7 +39,7 @@ func (s *Server) handleListMissingBlobs(e echo.Context) error { var records []models.Record if err := s.db.Raw(ctx, "SELECT * FROM records WHERE did = ?", nil, urepo.Repo.Did).Scan(&records).Error; err != nil { - s.logger.Error("failed to get records for listMissingBlobs", "error", err) + logger.Error("failed to get records for listMissingBlobs", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_list_records.go b/server/handle_repo_list_records.go index a2d7a61..c657465 100644 --- a/server/handle_repo_list_records.go +++ b/server/handle_repo_list_records.go @@ -47,10 +47,11 @@ func getLimitFromContext(e echo.Context, def int) (int, error) { func (s *Server) handleListRecords(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleListRecords") var req ComAtprotoRepoListRecordsRequest if err := e.Bind(&req); err != nil { - s.logger.Error("could not bind list records request", "error", err) + logger.Error("could not bind list records request", "error", err) return helpers.ServerError(e, nil) } @@ -96,7 +97,7 @@ func (s *Server) handleListRecords(e echo.Context) error { var records []models.Record if err := s.db.Raw(ctx, "SELECT * FROM records WHERE did = ? AND nsid = ? "+cursorquery+" ORDER BY created_at "+sort+" limit ?", nil, params...).Scan(&records).Error; err != nil { - s.logger.Error("error getting records", "error", err) + logger.Error("error getting records", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_put_record.go b/server/handle_repo_put_record.go index 12ec7ad..5c3e11e 100644 --- a/server/handle_repo_put_record.go +++ b/server/handle_repo_put_record.go @@ -18,22 +18,23 @@ type ComAtprotoRepoPutRecordInput struct { func (s *Server) handlePutRecord(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handlePutRecord") repo := e.Get("repo").(*models.RepoActor) var req ComAtprotoRepoPutRecordInput if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.InputError(e, nil) } if repo.Repo.Did != req.Repo { - s.logger.Warn("mismatched repo/auth") + logger.Warn("mismatched repo/auth") return helpers.InputError(e, nil) } @@ -53,7 +54,7 @@ func (s *Server) handlePutRecord(e echo.Context) error { }, }, req.SwapCommit) if err != nil { - s.logger.Error("error applying writes", "error", err) + logger.Error("error applying writes", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_upload_blob.go b/server/handle_repo_upload_blob.go index 3d863a7..d3f3dac 100644 --- a/server/handle_repo_upload_blob.go +++ b/server/handle_repo_upload_blob.go @@ -33,6 +33,7 @@ type ComAtprotoRepoUploadBlobResponse struct { func (s *Server) handleRepoUploadBlob(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleRepoUploadBlob") urepo := e.Get("repo").(*models.RepoActor) @@ -54,7 +55,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { } if err := s.db.Create(ctx, &blob, nil).Error; err != nil { - s.logger.Error("error creating new blob in db", "error", err) + logger.Error("error creating new blob in db", "error", err) return helpers.ServerError(e, nil) } @@ -71,7 +72,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { break } } else if err != nil && err != io.ErrUnexpectedEOF { - s.logger.Error("error reading blob", "error", err) + logger.Error("error reading blob", "error", err) return helpers.ServerError(e, nil) } @@ -87,7 +88,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { } if err := s.db.Create(ctx, &blobPart, nil).Error; err != nil { - s.logger.Error("error adding blob part to db", "error", err) + logger.Error("error adding blob part to db", "error", err) return helpers.ServerError(e, nil) } } @@ -100,7 +101,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { c, err := cid.NewPrefixV1(cid.Raw, multihash.SHA2_256).Sum(fulldata.Bytes()) if err != nil { - s.logger.Error("error creating cid prefix", "error", err) + logger.Error("error creating cid prefix", "error", err) return helpers.ServerError(e, nil) } @@ -117,7 +118,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { sess, err := session.NewSession(config) if err != nil { - s.logger.Error("error creating aws session", "error", err) + logger.Error("error creating aws session", "error", err) return helpers.ServerError(e, nil) } @@ -128,14 +129,14 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { Key: aws.String(fmt.Sprintf("blobs/%s/%s", urepo.Repo.Did, c.String())), Body: bytes.NewReader(fulldata.Bytes()), }); err != nil { - s.logger.Error("error uploading blob to s3", "error", err) + logger.Error("error uploading blob to s3", "error", err) return helpers.ServerError(e, nil) } } if err := s.db.Exec(ctx, "UPDATE blobs SET cid = ? WHERE id = ?", nil, c.Bytes(), blob.ID).Error; err != nil { // there should probably be somme handling here if this fails... - s.logger.Error("error updating blob", "error", err) + logger.Error("error updating blob", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_activate_account.go b/server/handle_server_activate_account.go index c5d2106..7db1aa2 100644 --- a/server/handle_server_activate_account.go +++ b/server/handle_server_activate_account.go @@ -19,17 +19,18 @@ type ComAtprotoServerActivateAccountRequest struct { func (s *Server) handleServerActivateAccount(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerActivateAccount") var req ComAtprotoServerDeactivateAccountRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } urepo := e.Get("repo").(*models.RepoActor) if err := s.db.Exec(ctx, "UPDATE repos SET deactivated = ? WHERE did = ?", nil, false, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating account status to deactivated", "error", err) + logger.Error("error updating account status to deactivated", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_check_account_status.go b/server/handle_server_check_account_status.go index e9a76f2..5092451 100644 --- a/server/handle_server_check_account_status.go +++ b/server/handle_server_check_account_status.go @@ -21,6 +21,7 @@ type ComAtprotoServerCheckAccountStatusResponse struct { func (s *Server) handleServerCheckAccountStatus(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerCheckAccountStatus") urepo := e.Get("repo").(*models.RepoActor) @@ -33,7 +34,7 @@ func (s *Server) handleServerCheckAccountStatus(e echo.Context) error { rootcid, err := cid.Cast(urepo.Root) if err != nil { - s.logger.Error("error casting cid", "error", err) + logger.Error("error casting cid", "error", err) return helpers.ServerError(e, nil) } resp.RepoCommit = rootcid.String() @@ -44,21 +45,21 @@ func (s *Server) handleServerCheckAccountStatus(e echo.Context) error { var blockCtResp CountResp if err := s.db.Raw(ctx, "SELECT COUNT(*) AS ct FROM blocks WHERE did = ?", nil, urepo.Repo.Did).Scan(&blockCtResp).Error; err != nil { - s.logger.Error("error getting block count", "error", err) + logger.Error("error getting block count", "error", err) return helpers.ServerError(e, nil) } resp.RepoBlocks = blockCtResp.Ct var recCtResp CountResp if err := s.db.Raw(ctx, "SELECT COUNT(*) AS ct FROM records WHERE did = ?", nil, urepo.Repo.Did).Scan(&recCtResp).Error; err != nil { - s.logger.Error("error getting record count", "error", err) + logger.Error("error getting record count", "error", err) return helpers.ServerError(e, nil) } resp.IndexedRecords = recCtResp.Ct var blobCtResp CountResp if err := s.db.Raw(ctx, "SELECT COUNT(*) AS ct FROM blobs WHERE did = ?", nil, urepo.Repo.Did).Scan(&blobCtResp).Error; err != nil { - s.logger.Error("error getting record count", "error", err) + logger.Error("error getting record count", "error", err) return helpers.ServerError(e, nil) } resp.ExpectedBlobs = blobCtResp.Ct diff --git a/server/handle_server_confirm_email.go b/server/handle_server_confirm_email.go index 5ab241a..895edc0 100644 --- a/server/handle_server_confirm_email.go +++ b/server/handle_server_confirm_email.go @@ -16,12 +16,13 @@ type ComAtprotoServerConfirmEmailRequest struct { func (s *Server) handleServerConfirmEmail(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerConfirmEmail") urepo := e.Get("repo").(*models.RepoActor) var req ComAtprotoServerConfirmEmailRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } @@ -44,7 +45,7 @@ func (s *Server) handleServerConfirmEmail(e echo.Context) error { now := time.Now().UTC() if err := s.db.Exec(ctx, "UPDATE repos SET email_verification_code = NULL, email_verification_code_expires_at = NULL, email_confirmed_at = ? WHERE did = ?", nil, now, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating user", "error", err) + logger.Error("error updating user", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_account.go b/server/handle_server_create_account.go index c0916df..a6f11aa 100644 --- a/server/handle_server_create_account.go +++ b/server/handle_server_create_account.go @@ -37,18 +37,19 @@ type ComAtprotoServerCreateAccountResponse struct { func (s *Server) handleCreateAccount(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerCreateAccount") var request ComAtprotoServerCreateAccountRequest if err := e.Bind(&request); err != nil { - s.logger.Error("error receiving request", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error receiving request", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.ServerError(e, nil) } request.Handle = strings.ToLower(request.Handle) if err := e.Validate(request); err != nil { - s.logger.Error("error validating request", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error validating request", "endpoint", "com.atproto.server.createAccount", "error", err) var verr ValidationError if errors.As(err, &verr) { @@ -82,7 +83,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { authDid, err := s.validateServiceAuth(e.Request().Context(), token, "com.atproto.server.createAccount") if err != nil { - s.logger.Warn("error validating authorization token", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Warn("error validating authorization token", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.UnauthorizedError(e, to.StringPtr("invalid authorization token")) } @@ -94,7 +95,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { // see if the handle is already taken actor, err := s.getActorByHandle(ctx, request.Handle) if err != nil && err != gorm.ErrRecordNotFound { - s.logger.Error("error looking up handle in db", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error looking up handle in db", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.ServerError(e, nil) } if err == nil && actor.Did != signupDid { @@ -115,7 +116,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { if err == gorm.ErrRecordNotFound { return helpers.InputError(e, to.StringPtr("InvalidInviteCode")) } - s.logger.Error("error getting invite code from db", "error", err) + logger.Error("error getting invite code from db", "error", err) return helpers.ServerError(e, nil) } @@ -127,7 +128,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { // see if the email is already taken existingRepo, err := s.getRepoByEmail(ctx, request.Email) if err != nil && err != gorm.ErrRecordNotFound { - s.logger.Error("error looking up email in db", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error looking up email in db", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.ServerError(e, nil) } if err == nil && existingRepo.Did != signupDid { @@ -141,17 +142,17 @@ func (s *Server) handleCreateAccount(e echo.Context) error { if signupDid != "" { reservedKey, err := s.getReservedKey(ctx, signupDid) if err != nil { - s.logger.Error("error looking up reserved key", "error", err) + logger.Error("error looking up reserved key", "error", err) } if reservedKey != nil { k, err = atcrypto.ParsePrivateBytesK256(reservedKey.PrivateKey) if err != nil { - s.logger.Error("error parsing reserved key", "error", err) + logger.Error("error parsing reserved key", "error", err) k = nil } else { defer func() { if delErr := s.deleteReservedKey(ctx, reservedKey.KeyDid, reservedKey.Did); delErr != nil { - s.logger.Error("error deleting reserved key", "error", delErr) + logger.Error("error deleting reserved key", "error", delErr) } }() } @@ -161,7 +162,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { if k == nil { k, err = atcrypto.GeneratePrivateKeyK256() if err != nil { - s.logger.Error("error creating signing key", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error creating signing key", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.ServerError(e, nil) } } @@ -169,12 +170,12 @@ func (s *Server) handleCreateAccount(e echo.Context) error { if signupDid == "" { did, op, err := s.plcClient.CreateDID(k, "", request.Handle) if err != nil { - s.logger.Error("error creating operation", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error creating operation", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.ServerError(e, nil) } if err := s.plcClient.SendOperation(e.Request().Context(), did, op); err != nil { - s.logger.Error("error sending plc op", "endpoint", "com.atproto.server.createAccount", "error", err) + logger.Error("error sending plc op", "endpoint", "com.atproto.server.createAccount", "error", err) return helpers.ServerError(e, nil) } signupDid = did @@ -182,7 +183,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { hashed, err := bcrypt.GenerateFromPassword([]byte(request.Password), 10) if err != nil { - s.logger.Error("error hashing password", "error", err) + logger.Error("error hashing password", "error", err) return helpers.ServerError(e, nil) } @@ -202,17 +203,17 @@ func (s *Server) handleCreateAccount(e echo.Context) error { } if err := s.db.Create(ctx, &urepo, nil).Error; err != nil { - s.logger.Error("error inserting new repo", "error", err) + logger.Error("error inserting new repo", "error", err) return helpers.ServerError(e, nil) } if err := s.db.Create(ctx, &actor, nil).Error; err != nil { - s.logger.Error("error inserting new actor", "error", err) + logger.Error("error inserting new actor", "error", err) return helpers.ServerError(e, nil) } } else { if err := s.db.Save(ctx, &actor, nil).Error; err != nil { - s.logger.Error("error inserting new actor", "error", err) + logger.Error("error inserting new actor", "error", err) return helpers.ServerError(e, nil) } } @@ -223,12 +224,12 @@ func (s *Server) handleCreateAccount(e echo.Context) error { root, rev, err := r.Commit(context.TODO(), urepo.SignFor) if err != nil { - s.logger.Error("error committing", "error", err) + logger.Error("error committing", "error", err) return helpers.ServerError(e, nil) } if err := s.UpdateRepo(context.TODO(), urepo.Did, root, rev); err != nil { - s.logger.Error("error updating repo after commit", "error", err) + logger.Error("error updating repo after commit", "error", err) return helpers.ServerError(e, nil) } @@ -244,23 +245,23 @@ func (s *Server) handleCreateAccount(e echo.Context) error { if s.config.RequireInvite { if err := s.db.Raw(ctx, "UPDATE invite_codes SET remaining_use_count = remaining_use_count - 1 WHERE code = ?", nil, request.InviteCode).Scan(&ic).Error; err != nil { - s.logger.Error("error decrementing use count", "error", err) + logger.Error("error decrementing use count", "error", err) return helpers.ServerError(e, nil) } } sess, err := s.createSession(ctx, &urepo) if err != nil { - s.logger.Error("error creating new session", "error", err) + logger.Error("error creating new session", "error", err) return helpers.ServerError(e, nil) } go func() { if err := s.sendEmailVerification(urepo.Email, actor.Handle, *urepo.EmailVerificationCode); err != nil { - s.logger.Error("error sending email verification email", "error", err) + logger.Error("error sending email verification email", "error", err) } if err := s.sendWelcomeMail(urepo.Email, actor.Handle); err != nil { - s.logger.Error("error sending welcome email", "error", err) + logger.Error("error sending welcome email", "error", err) } }() diff --git a/server/handle_server_create_invite_code.go b/server/handle_server_create_invite_code.go index 5a77e83..9909688 100644 --- a/server/handle_server_create_invite_code.go +++ b/server/handle_server_create_invite_code.go @@ -18,15 +18,16 @@ type ComAtprotoServerCreateInviteCodeResponse struct { func (s *Server) handleCreateInviteCode(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerCreateInviteCode") var req ComAtprotoServerCreateInviteCodeRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.InputError(e, nil) } @@ -44,7 +45,7 @@ func (s *Server) handleCreateInviteCode(e echo.Context) error { Did: acc, RemainingUseCount: req.UseCount, }, nil).Error; err != nil { - s.logger.Error("error creating invite code", "error", err) + logger.Error("error creating invite code", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_invite_codes.go b/server/handle_server_create_invite_codes.go index e75e704..42eb952 100644 --- a/server/handle_server_create_invite_codes.go +++ b/server/handle_server_create_invite_codes.go @@ -23,15 +23,16 @@ type ComAtprotoServerCreateInviteCodesItem struct { func (s *Server) handleCreateInviteCodes(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerCreateInviteCodes") var req ComAtprotoServerCreateInviteCodesRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.InputError(e, nil) } @@ -57,7 +58,7 @@ func (s *Server) handleCreateInviteCodes(e echo.Context) error { Did: did, RemainingUseCount: req.UseCount, }, nil).Error; err != nil { - s.logger.Error("error creating invite code", "error", err) + logger.Error("error creating invite code", "error", err) return helpers.ServerError(e, nil) } } diff --git a/server/handle_server_create_session.go b/server/handle_server_create_session.go index 645192d..fa9a888 100644 --- a/server/handle_server_create_session.go +++ b/server/handle_server_create_session.go @@ -33,10 +33,11 @@ type ComAtprotoServerCreateSessionResponse struct { func (s *Server) handleCreateSession(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerCreateSession") var req ComAtprotoServerCreateSessionRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding request", "endpoint", "com.atproto.server.serverCreateSession", "error", err) + logger.Error("error binding request", "endpoint", "com.atproto.server.serverCreateSession", "error", err) return helpers.ServerError(e, nil) } @@ -79,20 +80,20 @@ func (s *Server) handleCreateSession(e echo.Context) error { return helpers.InputError(e, to.StringPtr("InvalidRequest")) } - s.logger.Error("erorr looking up repo", "endpoint", "com.atproto.server.createSession", "error", err) + logger.Error("erorr looking up repo", "endpoint", "com.atproto.server.createSession", "error", err) return helpers.ServerError(e, nil) } if err := bcrypt.CompareHashAndPassword([]byte(repo.Password), []byte(req.Password)); err != nil { if err != bcrypt.ErrMismatchedHashAndPassword { - s.logger.Error("erorr comparing hash and password", "error", err) + logger.Error("erorr comparing hash and password", "error", err) } return helpers.InputError(e, to.StringPtr("InvalidRequest")) } sess, err := s.createSession(ctx, &repo.Repo) if err != nil { - s.logger.Error("error creating session", "error", err) + logger.Error("error creating session", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_deactivate_account.go b/server/handle_server_deactivate_account.go index 02db068..e027af2 100644 --- a/server/handle_server_deactivate_account.go +++ b/server/handle_server_deactivate_account.go @@ -20,17 +20,18 @@ type ComAtprotoServerDeactivateAccountRequest struct { func (s *Server) handleServerDeactivateAccount(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerDeactivateAccount") var req ComAtprotoServerDeactivateAccountRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } urepo := e.Get("repo").(*models.RepoActor) if err := s.db.Exec(ctx, "UPDATE repos SET deactivated = ? WHERE did = ?", nil, true, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating account status to deactivated", "error", err) + logger.Error("error updating account status to deactivated", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_delete_account.go b/server/handle_server_delete_account.go index 52c4e0a..1336b70 100644 --- a/server/handle_server_delete_account.go +++ b/server/handle_server_delete_account.go @@ -21,31 +21,32 @@ type ComAtprotoServerDeleteAccountRequest struct { func (s *Server) handleServerDeleteAccount(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerDeleteAccount") var req ComAtprotoServerDeleteAccountRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } if err := e.Validate(&req); err != nil { - s.logger.Error("error validating", "error", err) + logger.Error("error validating", "error", err) return helpers.ServerError(e, nil) } urepo, err := s.getRepoActorByDid(ctx, req.Did) if err != nil { - s.logger.Error("error getting repo", "error", err) + logger.Error("error getting repo", "error", err) return echo.NewHTTPError(400, "account not found") } if err := bcrypt.CompareHashAndPassword([]byte(urepo.Repo.Password), []byte(req.Password)); err != nil { - s.logger.Error("password mismatch", "error", err) + logger.Error("password mismatch", "error", err) return echo.NewHTTPError(401, "Invalid did or password") } if urepo.Repo.AccountDeleteCode == nil || urepo.Repo.AccountDeleteCodeExpiresAt == nil { - s.logger.Error("no deletion token found for account") + logger.Error("no deletion token found for account") return echo.NewHTTPError(400, map[string]interface{}{ "error": "InvalidToken", "message": "Token is invalid", @@ -53,7 +54,7 @@ func (s *Server) handleServerDeleteAccount(e echo.Context) error { } if *urepo.Repo.AccountDeleteCode != req.Token { - s.logger.Error("deletion token mismatch") + logger.Error("deletion token mismatch") return echo.NewHTTPError(400, map[string]interface{}{ "error": "InvalidToken", "message": "Token is invalid", @@ -61,7 +62,7 @@ func (s *Server) handleServerDeleteAccount(e echo.Context) error { } if time.Now().UTC().After(*urepo.Repo.AccountDeleteCodeExpiresAt) { - s.logger.Error("deletion token expired") + logger.Error("deletion token expired") return echo.NewHTTPError(400, map[string]interface{}{ "error": "ExpiredToken", "message": "Token is expired", @@ -70,66 +71,68 @@ func (s *Server) handleServerDeleteAccount(e echo.Context) error { tx := s.db.BeginDangerously(ctx) if tx.Error != nil { - s.logger.Error("error starting transaction", "error", tx.Error) + logger.Error("error starting transaction", "error", tx.Error) return helpers.ServerError(e, nil) } + status := "error" + func() { + if status == "error" { + if err := tx.Rollback().Error; err != nil { + logger.Error("error rolling back after delete failure", "err", err) + } + } + }() + if err := tx.Exec("DELETE FROM blocks WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting blocks", "error", err) + logger.Error("error deleting blocks", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM records WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting records", "error", err) + logger.Error("error deleting records", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM blobs WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting blobs", "error", err) + logger.Error("error deleting blobs", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM tokens WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting tokens", "error", err) + logger.Error("error deleting tokens", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM refresh_tokens WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting refresh tokens", "error", err) + logger.Error("error deleting refresh tokens", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM reserved_keys WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting reserved keys", "error", err) + logger.Error("error deleting reserved keys", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM invite_codes WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting invite codes", "error", err) + logger.Error("error deleting invite codes", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM actors WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting actor", "error", err) + logger.Error("error deleting actor", "error", err) return helpers.ServerError(e, nil) } if err := tx.Exec("DELETE FROM repos WHERE did = ?", nil, req.Did).Error; err != nil { - tx.Rollback() - s.logger.Error("error deleting repo", "error", err) + logger.Error("error deleting repo", "error", err) return helpers.ServerError(e, nil) } + status = "ok" + if err := tx.Commit().Error; err != nil { - s.logger.Error("error committing transaction", "error", err) + logger.Error("error committing transaction", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_get_service_auth.go b/server/handle_server_get_service_auth.go index 3a4b07a..969afb3 100644 --- a/server/handle_server_get_service_auth.go +++ b/server/handle_server_get_service_auth.go @@ -25,9 +25,11 @@ type ServerGetServiceAuthRequest struct { } func (s *Server) handleServerGetServiceAuth(e echo.Context) error { + logger := s.logger.With("name", "handleServerGetServiceAuth") + var req ServerGetServiceAuthRequest if err := e.Bind(&req); err != nil { - s.logger.Error("could not bind service auth request", "error", err) + logger.Error("could not bind service auth request", "error", err) return helpers.ServerError(e, nil) } @@ -64,7 +66,7 @@ func (s *Server) handleServerGetServiceAuth(e echo.Context) error { } hj, err := json.Marshal(header) if err != nil { - s.logger.Error("error marshaling header", "error", err) + logger.Error("error marshaling header", "error", err) return helpers.ServerError(e, nil) } @@ -82,7 +84,7 @@ func (s *Server) handleServerGetServiceAuth(e echo.Context) error { } pj, err := json.Marshal(payload) if err != nil { - s.logger.Error("error marashaling payload", "error", err) + logger.Error("error marashaling payload", "error", err) return helpers.ServerError(e, nil) } @@ -93,13 +95,13 @@ func (s *Server) handleServerGetServiceAuth(e echo.Context) error { sk, err := secp256k1secec.NewPrivateKey(repo.SigningKey) if err != nil { - s.logger.Error("can't load private key", "error", err) + logger.Error("can't load private key", "error", err) return err } R, S, _, err := sk.SignRaw(rand.Reader, hash[:]) if err != nil { - s.logger.Error("error signing", "error", err) + logger.Error("error signing", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_refresh_session.go b/server/handle_server_refresh_session.go index 34110b6..c3efc4a 100644 --- a/server/handle_server_refresh_session.go +++ b/server/handle_server_refresh_session.go @@ -17,23 +17,24 @@ type ComAtprotoServerRefreshSessionResponse struct { func (s *Server) handleRefreshSession(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerRefreshSession") token := e.Get("token").(string) repo := e.Get("repo").(*models.RepoActor) if err := s.db.Exec(ctx, "DELETE FROM refresh_tokens WHERE token = ?", nil, token).Error; err != nil { - s.logger.Error("error getting refresh token from db", "error", err) + logger.Error("error getting refresh token from db", "error", err) return helpers.ServerError(e, nil) } if err := s.db.Exec(ctx, "DELETE FROM tokens WHERE refresh_token = ?", nil, token).Error; err != nil { - s.logger.Error("error deleting access token from db", "error", err) + logger.Error("error deleting access token from db", "error", err) return helpers.ServerError(e, nil) } sess, err := s.createSession(ctx, &repo.Repo) if err != nil { - s.logger.Error("error creating new session for refresh", "error", err) + logger.Error("error creating new session for refresh", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_request_account_delete.go b/server/handle_server_request_account_delete.go index ea8f8aa..53494e7 100644 --- a/server/handle_server_request_account_delete.go +++ b/server/handle_server_request_account_delete.go @@ -11,6 +11,7 @@ import ( func (s *Server) handleServerRequestAccountDelete(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerRequestAccountDelete") urepo := e.Get("repo").(*models.RepoActor) @@ -18,13 +19,13 @@ func (s *Server) handleServerRequestAccountDelete(e echo.Context) error { expiresAt := time.Now().UTC().Add(15 * time.Minute) if err := s.db.Exec(ctx, "UPDATE repos SET account_delete_code = ?, account_delete_code_expires_at = ? WHERE did = ?", nil, token, expiresAt, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error setting deletion token", "error", err) + logger.Error("error setting deletion token", "error", err) return helpers.ServerError(e, nil) } if urepo.Email != "" { if err := s.sendAccountDeleteEmail(urepo.Email, urepo.Actor.Handle, token); err != nil { - s.logger.Error("error sending account deletion email", "error", err) + logger.Error("error sending account deletion email", "error", err) } } diff --git a/server/handle_server_request_email_confirmation.go b/server/handle_server_request_email_confirmation.go index dce9981..4dcac8c 100644 --- a/server/handle_server_request_email_confirmation.go +++ b/server/handle_server_request_email_confirmation.go @@ -12,6 +12,7 @@ import ( func (s *Server) handleServerRequestEmailConfirmation(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerRequestEmailConfirm") urepo := e.Get("repo").(*models.RepoActor) @@ -23,12 +24,12 @@ func (s *Server) handleServerRequestEmailConfirmation(e echo.Context) error { eat := time.Now().Add(10 * time.Minute).UTC() if err := s.db.Exec(ctx, "UPDATE repos SET email_verification_code = ?, email_verification_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating user", "error", err) + logger.Error("error updating user", "error", err) return helpers.ServerError(e, nil) } if err := s.sendEmailVerification(urepo.Email, urepo.Handle, code); err != nil { - s.logger.Error("error sending mail", "error", err) + logger.Error("error sending mail", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_request_email_update.go b/server/handle_server_request_email_update.go index 7ab4b48..85e8efe 100644 --- a/server/handle_server_request_email_update.go +++ b/server/handle_server_request_email_update.go @@ -15,6 +15,7 @@ type ComAtprotoRequestEmailUpdateResponse struct { func (s *Server) handleServerRequestEmailUpdate(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerRequestEmailUpdate") urepo := e.Get("repo").(*models.RepoActor) @@ -23,12 +24,12 @@ func (s *Server) handleServerRequestEmailUpdate(e echo.Context) error { eat := time.Now().Add(10 * time.Minute).UTC() if err := s.db.Exec(ctx, "UPDATE repos SET email_update_code = ?, email_update_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating repo", "error", err) + logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } if err := s.sendEmailUpdate(urepo.Email, urepo.Handle, code); err != nil { - s.logger.Error("error sending email", "error", err) + logger.Error("error sending email", "error", err) return helpers.ServerError(e, nil) } } diff --git a/server/handle_server_request_password_reset.go b/server/handle_server_request_password_reset.go index 2d90451..6149ee2 100644 --- a/server/handle_server_request_password_reset.go +++ b/server/handle_server_request_password_reset.go @@ -15,6 +15,7 @@ type ComAtprotoServerRequestPasswordResetRequest struct { func (s *Server) handleServerRequestPasswordReset(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerRequestPasswordReset") urepo, ok := e.Get("repo").(*models.RepoActor) if !ok { @@ -39,12 +40,12 @@ func (s *Server) handleServerRequestPasswordReset(e echo.Context) error { eat := time.Now().Add(10 * time.Minute).UTC() if err := s.db.Exec(ctx, "UPDATE repos SET password_reset_code = ?, password_reset_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating repo", "error", err) + logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } if err := s.sendPasswordReset(urepo.Email, urepo.Handle, code); err != nil { - s.logger.Error("error sending email", "error", err) + logger.Error("error sending email", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_reserve_signing_key.go b/server/handle_server_reserve_signing_key.go index 036be81..d35d1c1 100644 --- a/server/handle_server_reserve_signing_key.go +++ b/server/handle_server_reserve_signing_key.go @@ -20,10 +20,11 @@ type ServerReserveSigningKeyResponse struct { func (s *Server) handleServerReserveSigningKey(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerReserveSigningKey") var req ServerReserveSigningKeyRequest if err := e.Bind(&req); err != nil { - s.logger.Error("could not bind reserve signing key request", "error", err) + logger.Error("could not bind reserve signing key request", "error", err) return helpers.ServerError(e, nil) } @@ -38,13 +39,13 @@ func (s *Server) handleServerReserveSigningKey(e echo.Context) error { k, err := atcrypto.GeneratePrivateKeyK256() if err != nil { - s.logger.Error("error creating signing key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err) + logger.Error("error creating signing key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err) return helpers.ServerError(e, nil) } pubKey, err := k.PublicKey() if err != nil { - s.logger.Error("error getting public key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err) + logger.Error("error getting public key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err) return helpers.ServerError(e, nil) } @@ -58,11 +59,11 @@ func (s *Server) handleServerReserveSigningKey(e echo.Context) error { } if err := s.db.Create(ctx, &reservedKey, nil).Error; err != nil { - s.logger.Error("error storing reserved key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err) + logger.Error("error storing reserved key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err) return helpers.ServerError(e, nil) } - s.logger.Info("reserved signing key", "keyDid", keyDid, "forDid", req.Did) + logger.Info("reserved signing key", "keyDid", keyDid, "forDid", req.Did) return e.JSON(200, ServerReserveSigningKeyResponse{ SigningKey: keyDid, diff --git a/server/handle_server_reset_password.go b/server/handle_server_reset_password.go index 473b5de..6fd104d 100644 --- a/server/handle_server_reset_password.go +++ b/server/handle_server_reset_password.go @@ -17,12 +17,13 @@ type ComAtprotoServerResetPasswordRequest struct { func (s *Server) handleServerResetPassword(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerResetPassword") urepo := e.Get("repo").(*models.RepoActor) var req ComAtprotoServerResetPasswordRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } @@ -44,12 +45,12 @@ func (s *Server) handleServerResetPassword(e echo.Context) error { hash, err := bcrypt.GenerateFromPassword([]byte(req.Password), 10) if err != nil { - s.logger.Error("error creating hash", "error", err) + logger.Error("error creating hash", "error", err) return helpers.ServerError(e, nil) } if err := s.db.Exec(ctx, "UPDATE repos SET password_reset_code = NULL, password_reset_code_expires_at = NULL, password = ? WHERE did = ?", nil, hash, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating repo", "error", err) + logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_resolve_handle.go b/server/handle_server_resolve_handle.go index 111745f..51c4bf6 100644 --- a/server/handle_server_resolve_handle.go +++ b/server/handle_server_resolve_handle.go @@ -10,6 +10,8 @@ import ( ) func (s *Server) handleResolveHandle(e echo.Context) error { + logger := s.logger.With("name", "handleServerResolveHandle") + type Resp struct { Did string `json:"did"` } @@ -28,7 +30,7 @@ func (s *Server) handleResolveHandle(e echo.Context) error { ctx := context.WithValue(e.Request().Context(), "skip-cache", true) did, err := s.passport.ResolveHandle(ctx, parsed.String()) if err != nil { - s.logger.Error("error resolving handle", "error", err) + logger.Error("error resolving handle", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_update_email.go b/server/handle_server_update_email.go index c94e86a..7d23f00 100644 --- a/server/handle_server_update_email.go +++ b/server/handle_server_update_email.go @@ -16,12 +16,13 @@ type ComAtprotoServerUpdateEmailRequest struct { func (s *Server) handleServerUpdateEmail(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleServerUpdateEmail") urepo := e.Get("repo").(*models.RepoActor) var req ComAtprotoServerUpdateEmailRequest if err := e.Bind(&req); err != nil { - s.logger.Error("error binding", "error", err) + logger.Error("error binding", "error", err) return helpers.ServerError(e, nil) } @@ -42,7 +43,7 @@ func (s *Server) handleServerUpdateEmail(e echo.Context) error { } if err := s.db.Exec(ctx, "UPDATE repos SET email_update_code = NULL, email_update_code_expires_at = NULL, email_confirmed_at = NULL, email = ? WHERE did = ?", nil, req.Email, urepo.Repo.Did).Error; err != nil { - s.logger.Error("error updating repo", "error", err) + logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_get_blob.go b/server/handle_sync_get_blob.go index e136d5a..8bddc17 100644 --- a/server/handle_sync_get_blob.go +++ b/server/handle_sync_get_blob.go @@ -18,6 +18,7 @@ import ( func (s *Server) handleSyncGetBlob(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleSyncGetBlob") did := e.QueryParam("did") if did == "" { @@ -36,7 +37,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { urepo, err := s.getRepoActorByDid(ctx, did) if err != nil { - s.logger.Error("could not find user for requested blob", "error", err) + logger.Error("could not find user for requested blob", "error", err) return helpers.InputError(e, nil) } @@ -49,7 +50,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { var blob models.Blob if err := s.db.Raw(ctx, "SELECT * FROM blobs WHERE did = ? AND cid = ?", nil, did, c.Bytes()).Scan(&blob).Error; err != nil { - s.logger.Error("error looking up blob", "error", err) + logger.Error("error looking up blob", "error", err) return helpers.ServerError(e, nil) } @@ -58,7 +59,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { if blob.Storage == "sqlite" { var parts []models.BlobPart if err := s.db.Raw(ctx, "SELECT * FROM blob_parts WHERE blob_id = ? ORDER BY idx", nil, blob.ID).Scan(&parts).Error; err != nil { - s.logger.Error("error getting blob parts", "error", err) + logger.Error("error getting blob parts", "error", err) return helpers.ServerError(e, nil) } @@ -68,7 +69,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { } } else if blob.Storage == "s3" { if !(s.s3Config != nil && s.s3Config.BlobstoreEnabled) { - s.logger.Error("s3 storage disabled") + logger.Error("s3 storage disabled") return helpers.ServerError(e, nil) } @@ -91,7 +92,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { sess, err := session.NewSession(config) if err != nil { - s.logger.Error("error creating aws session", "error", err) + logger.Error("error creating aws session", "error", err) return helpers.ServerError(e, nil) } @@ -100,7 +101,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { Bucket: aws.String(s.s3Config.Bucket), Key: aws.String(blobKey), }); err != nil { - s.logger.Error("error getting blob from s3", "error", err) + logger.Error("error getting blob from s3", "error", err) return helpers.ServerError(e, nil) } else { read := 0 @@ -114,7 +115,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { break } } else if err != nil && err != io.ErrUnexpectedEOF { - s.logger.Error("error reading blob", "error", err) + logger.Error("error reading blob", "error", err) return helpers.ServerError(e, nil) } @@ -125,7 +126,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { } } } else { - s.logger.Error("unknown storage", "storage", blob.Storage) + logger.Error("unknown storage", "storage", blob.Storage) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_get_blocks.go b/server/handle_sync_get_blocks.go index e9e0157..1577922 100644 --- a/server/handle_sync_get_blocks.go +++ b/server/handle_sync_get_blocks.go @@ -18,6 +18,7 @@ type ComAtprotoSyncGetBlocksRequest struct { func (s *Server) handleGetBlocks(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleSyncGetBlocks") var req ComAtprotoSyncGetBlocksRequest if err := e.Bind(&req); err != nil { @@ -52,7 +53,7 @@ func (s *Server) handleGetBlocks(e echo.Context) error { }) if _, err := carstore.LdWrite(buf, hb); err != nil { - s.logger.Error("error writing to car", "error", err) + logger.Error("error writing to car", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_get_record.go b/server/handle_sync_get_record.go index cc97595..613e476 100644 --- a/server/handle_sync_get_record.go +++ b/server/handle_sync_get_record.go @@ -14,6 +14,7 @@ import ( func (s *Server) handleSyncGetRecord(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleSyncGetRecord") did := e.QueryParam("did") collection := e.QueryParam("collection") @@ -21,7 +22,7 @@ func (s *Server) handleSyncGetRecord(e echo.Context) error { var urepo models.Repo if err := s.db.Raw(ctx, "SELECT * FROM repos WHERE did = ?", nil, did).Scan(&urepo).Error; err != nil { - s.logger.Error("error getting repo", "error", err) + logger.Error("error getting repo", "error", err) return helpers.ServerError(e, nil) } @@ -38,13 +39,13 @@ func (s *Server) handleSyncGetRecord(e echo.Context) error { }) if _, err := carstore.LdWrite(buf, hb); err != nil { - s.logger.Error("error writing to car", "error", err) + logger.Error("error writing to car", "error", err) return helpers.ServerError(e, nil) } for _, blk := range blocks { if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil { - s.logger.Error("error writing to car", "error", err) + logger.Error("error writing to car", "error", err) return helpers.ServerError(e, nil) } } diff --git a/server/handle_sync_get_repo.go b/server/handle_sync_get_repo.go index dacc02b..152f14a 100644 --- a/server/handle_sync_get_repo.go +++ b/server/handle_sync_get_repo.go @@ -14,6 +14,7 @@ import ( func (s *Server) handleSyncGetRepo(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleSyncGetRepo") did := e.QueryParam("did") if did == "" { @@ -38,7 +39,7 @@ func (s *Server) handleSyncGetRepo(e echo.Context) error { buf := new(bytes.Buffer) if _, err := carstore.LdWrite(buf, hb); err != nil { - s.logger.Error("error writing to car", "error", err) + logger.Error("error writing to car", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_list_blobs.go b/server/handle_sync_list_blobs.go index a5a0aca..3855fbb 100644 --- a/server/handle_sync_list_blobs.go +++ b/server/handle_sync_list_blobs.go @@ -15,6 +15,7 @@ type ComAtprotoSyncListBlobsResponse struct { func (s *Server) handleSyncListBlobs(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleSyncListBlobs") did := e.QueryParam("did") if did == "" { @@ -39,7 +40,7 @@ func (s *Server) handleSyncListBlobs(e echo.Context) error { urepo, err := s.getRepoActorByDid(ctx, did) if err != nil { - s.logger.Error("could not find user for requested blobs", "error", err) + logger.Error("could not find user for requested blobs", "error", err) return helpers.InputError(e, nil) } @@ -52,7 +53,7 @@ func (s *Server) handleSyncListBlobs(e echo.Context) error { var blobs []models.Blob if err := s.db.Raw(ctx, "SELECT * FROM blobs WHERE did = ? "+cursorquery+" ORDER BY created_at DESC LIMIT ?", nil, params...).Scan(&blobs).Error; err != nil { - s.logger.Error("error getting records", "error", err) + logger.Error("error getting records", "error", err) return helpers.ServerError(e, nil) } @@ -60,7 +61,7 @@ func (s *Server) handleSyncListBlobs(e echo.Context) error { for _, b := range blobs { c, err := cid.Cast(b.Cid) if err != nil { - s.logger.Error("error casting cid", "error", err) + logger.Error("error casting cid", "error", err) return helpers.ServerError(e, nil) } cstrs = append(cstrs, c.String()) diff --git a/server/handle_sync_subscribe_repos.go b/server/handle_sync_subscribe_repos.go index d955aa4..369155d 100644 --- a/server/handle_sync_subscribe_repos.go +++ b/server/handle_sync_subscribe_repos.go @@ -7,6 +7,7 @@ import ( "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/lex/util" "github.com/btcsuite/websocket" + "github.com/haileyok/cocoon/metrics" "github.com/labstack/echo/v4" ) @@ -24,6 +25,11 @@ func (s *Server) handleSyncSubscribeRepos(e echo.Context) error { logger = logger.With("ident", ident) logger.Info("new connection established") + metrics.RelaysConnected.WithLabelValues(ident).Inc() + defer func() { + metrics.RelaysConnected.WithLabelValues(ident).Dec() + }() + evts, cancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool { return true }, nil) @@ -34,53 +40,59 @@ func (s *Server) handleSyncSubscribeRepos(e echo.Context) error { header := events.EventHeader{Op: events.EvtKindMessage} for evt := range evts { - wc, err := conn.NextWriter(websocket.BinaryMessage) - if err != nil { - logger.Error("error writing message to relay", "err", err) - break - } - - if ctx.Err() != nil { - logger.Error("context error", "err", err) - break - } - - var obj util.CBOR - switch { - case evt.Error != nil: - header.Op = events.EvtKindErrorFrame - obj = evt.Error - case evt.RepoCommit != nil: - header.MsgType = "#commit" - obj = evt.RepoCommit - case evt.RepoIdentity != nil: - header.MsgType = "#identity" - obj = evt.RepoIdentity - case evt.RepoAccount != nil: - header.MsgType = "#account" - obj = evt.RepoAccount - case evt.RepoInfo != nil: - header.MsgType = "#info" - obj = evt.RepoInfo - default: - logger.Warn("unrecognized event kind") - return nil - } - - if err := header.MarshalCBOR(wc); err != nil { - logger.Error("failed to write header to relay", "err", err) - break - } - - if err := obj.MarshalCBOR(wc); err != nil { - logger.Error("failed to write event to relay", "err", err) - break - } - - if err := wc.Close(); err != nil { - logger.Error("failed to flush-close our event write", "err", err) - break - } + func() { + defer func() { + metrics.RelaySends.WithLabelValues(header.MsgType).Inc() + }() + + wc, err := conn.NextWriter(websocket.BinaryMessage) + if err != nil { + logger.Error("error writing message to relay", "err", err) + return + } + + if ctx.Err() != nil { + logger.Error("context error", "err", err) + return + } + + var obj util.CBOR + switch { + case evt.Error != nil: + header.Op = events.EvtKindErrorFrame + obj = evt.Error + case evt.RepoCommit != nil: + header.MsgType = "#commit" + obj = evt.RepoCommit + case evt.RepoIdentity != nil: + header.MsgType = "#identity" + obj = evt.RepoIdentity + case evt.RepoAccount != nil: + header.MsgType = "#account" + obj = evt.RepoAccount + case evt.RepoInfo != nil: + header.MsgType = "#info" + obj = evt.RepoInfo + default: + logger.Warn("unrecognized event kind") + return + } + + if err := header.MarshalCBOR(wc); err != nil { + logger.Error("failed to write header to relay", "err", err) + return + } + + if err := obj.MarshalCBOR(wc); err != nil { + logger.Error("failed to write event to relay", "err", err) + return + } + + if err := wc.Close(); err != nil { + logger.Error("failed to flush-close our event write", "err", err) + return + } + }() } // we should tell the relay to request a new crawl at this point if we got disconnected diff --git a/server/handle_well_known.go b/server/handle_well_known.go index cd4091a..a1c4456 100644 --- a/server/handle_well_known.go +++ b/server/handle_well_known.go @@ -68,6 +68,7 @@ func (s *Server) handleWellKnown(e echo.Context) error { func (s *Server) handleAtprotoDid(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleAtprotoDid") host := e.Request().Host if host == "" { @@ -91,7 +92,7 @@ func (s *Server) handleAtprotoDid(e echo.Context) error { if err == gorm.ErrRecordNotFound { return e.NoContent(404) } - s.logger.Error("error looking up actor by handle", "error", err) + logger.Error("error looking up actor by handle", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/middleware.go b/server/middleware.go index 70e4ecf..1e116d8 100644 --- a/server/middleware.go +++ b/server/middleware.go @@ -38,6 +38,7 @@ func (s *Server) handleAdminMiddleware(next echo.HandlerFunc) echo.HandlerFunc { func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.HandlerFunc { return func(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleLegacySessionMiddleware") authheader := e.Request().Header.Get("authorization") if authheader == "" { @@ -69,20 +70,20 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl if hasLxm { pts := strings.Split(e.Request().URL.String(), "/") if lxm != pts[len(pts)-1] { - s.logger.Error("service auth lxm incorrect", "lxm", lxm, "expected", pts[len(pts)-1], "error", err) + logger.Error("service auth lxm incorrect", "lxm", lxm, "expected", pts[len(pts)-1], "error", err) return helpers.InputError(e, nil) } maybeDid, ok := claims["iss"].(string) if !ok { - s.logger.Error("no iss in service auth token", "error", err) + logger.Error("no iss in service auth token", "error", err) return helpers.InputError(e, nil) } did = maybeDid maybeRepo, err := s.getRepoActorByDid(ctx, did) if err != nil { - s.logger.Error("error fetching repo", "error", err) + logger.Error("error fetching repo", "error", err) return helpers.ServerError(e, nil) } repo = maybeRepo @@ -96,7 +97,7 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl return s.privateKey.Public(), nil }) if err != nil { - s.logger.Error("error parsing jwt", "error", err) + logger.Error("error parsing jwt", "error", err) return helpers.ExpiredTokenError(e) } @@ -109,12 +110,12 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl hash := sha256.Sum256([]byte(signingInput)) sigBytes, err := base64.RawURLEncoding.DecodeString(kpts[2]) if err != nil { - s.logger.Error("error decoding signature bytes", "error", err) + logger.Error("error decoding signature bytes", "error", err) return helpers.ServerError(e, nil) } if len(sigBytes) != 64 { - s.logger.Error("incorrect sigbytes length", "length", len(sigBytes)) + logger.Error("incorrect sigbytes length", "length", len(sigBytes)) return helpers.ServerError(e, nil) } @@ -140,19 +141,19 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl sk, err := secp256k1secec.NewPrivateKey(repo.SigningKey) if err != nil { - s.logger.Error("can't load private key", "error", err) + logger.Error("can't load private key", "error", err) return err } pubKey, ok := sk.Public().(*secp256k1secec.PublicKey) if !ok { - s.logger.Error("error getting public key from sk") + logger.Error("error getting public key from sk") return helpers.ServerError(e, nil) } verified := pubKey.VerifyRaw(hash[:], rr, ss) if !verified { - s.logger.Error("error verifying", "error", err) + logger.Error("error verifying", "error", err) return helpers.ServerError(e, nil) } } @@ -181,7 +182,7 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl return helpers.InvalidTokenError(e) } - s.logger.Error("error getting token from db", "error", err) + logger.Error("error getting token from db", "error", err) return helpers.ServerError(e, nil) } @@ -192,7 +193,7 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl exp, ok := claims["exp"].(float64) if !ok { - s.logger.Error("error getting iat from token") + logger.Error("error getting iat from token") return helpers.ServerError(e, nil) } @@ -203,7 +204,7 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl if repo == nil { maybeRepo, err := s.getRepoActorByDid(ctx, claims["sub"].(string)) if err != nil { - s.logger.Error("error fetching repo", "error", err) + logger.Error("error fetching repo", "error", err) return helpers.ServerError(e, nil) } repo = maybeRepo @@ -225,6 +226,7 @@ func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.Handl func (s *Server) handleOauthSessionMiddleware(next echo.HandlerFunc) echo.HandlerFunc { return func(e echo.Context) error { ctx := e.Request().Context() + logger := s.logger.With("name", "handleOauthSessionMiddleware") authheader := e.Request().Header.Get("authorization") if authheader == "" { @@ -257,13 +259,13 @@ func (s *Server) handleOauthSessionMiddleware(next echo.HandlerFunc) echo.Handle "error": "use_dpop_nonce", }) } - s.logger.Error("invalid dpop proof", "error", err) + logger.Error("invalid dpop proof", "error", err) return helpers.InputError(e, nil) } var oauthToken provider.OauthToken if err := s.db.Raw(ctx, "SELECT * FROM oauth_tokens WHERE token = ?", nil, accessToken).Scan(&oauthToken).Error; err != nil { - s.logger.Error("error finding access token in db", "error", err) + logger.Error("error finding access token in db", "error", err) return helpers.InputError(e, nil) } @@ -272,7 +274,7 @@ func (s *Server) handleOauthSessionMiddleware(next echo.HandlerFunc) echo.Handle } if *oauthToken.Parameters.DpopJkt != proof.JKT { - s.logger.Error("jkt mismatch", "token", oauthToken.Parameters.DpopJkt, "proof", proof.JKT) + logger.Error("jkt mismatch", "token", oauthToken.Parameters.DpopJkt, "proof", proof.JKT) return helpers.InputError(e, to.StringPtr("dpop jkt mismatch")) } @@ -287,7 +289,7 @@ func (s *Server) handleOauthSessionMiddleware(next echo.HandlerFunc) echo.Handle repo, err := s.getRepoActorByDid(ctx, oauthToken.Sub) if err != nil { - s.logger.Error("could not find actor in db", "error", err) + logger.Error("could not find actor in db", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/repo.go b/server/repo.go index 6d32e48..4dd6058 100644 --- a/server/repo.go +++ b/server/repo.go @@ -17,6 +17,7 @@ import ( lexutil "github.com/bluesky-social/indigo/lex/util" "github.com/bluesky-social/indigo/repo" "github.com/haileyok/cocoon/internal/db" + "github.com/haileyok/cocoon/metrics" "github.com/haileyok/cocoon/models" "github.com/haileyok/cocoon/recording_blockstore" blocks "github.com/ipfs/go-block-format" @@ -251,6 +252,12 @@ func (rm *RepoMan) applyWrites(ctx context.Context, urepo models.Repo, writes [] return nil, err } + for _, result := range results { + if result.Type != nil { + metrics.RepoOperations.WithLabelValues(*result.Type).Inc() + } + } + // create a buffer for dumping our new cbor into buf := new(bytes.Buffer) diff --git a/server/server.go b/server/server.go index 59ddc35..2ad717b 100644 --- a/server/server.go +++ b/server/server.go @@ -39,6 +39,7 @@ import ( "github.com/haileyok/cocoon/oauth/provider" "github.com/haileyok/cocoon/plc" "github.com/ipfs/go-cid" + "github.com/labstack/echo-contrib/echoprometheus" echo_session "github.com/labstack/echo-contrib/session" "github.com/labstack/echo/v4" "github.com/labstack/echo/v4/middleware" @@ -89,11 +90,12 @@ type Server struct { } type Args struct { + Logger *slog.Logger + Addr string DbName string DbType string DatabaseURL string - Logger *slog.Logger Version string Did string Hostname string @@ -209,6 +211,12 @@ func (t *TemplateRenderer) Render(w io.Writer, name string, data any, c echo.Con } func New(args *Args) (*Server, error) { + if args.Logger == nil { + args.Logger = slog.Default() + } + + logger := args.Logger.With("name", "New") + if args.Addr == "" { return nil, fmt.Errorf("addr must be set") } @@ -237,10 +245,6 @@ func New(args *Args) (*Server, error) { return nil, fmt.Errorf("admin password must be set") } - if args.Logger == nil { - args.Logger = slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{})) - } - if args.SessionSecret == "" { panic("SESSION SECRET WAS NOT SET. THIS IS REQUIRED. ") } @@ -248,8 +252,9 @@ func New(args *Args) (*Server, error) { e := echo.New() e.Pre(middleware.RemoveTrailingSlash()) - e.Pre(slogecho.New(args.Logger)) + e.Pre(slogecho.New(args.Logger.With("component", "slogecho"))) e.Use(echo_session.Middleware(sessions.NewCookieStore([]byte(args.SessionSecret)))) + e.Use(echoprometheus.NewMiddleware("cocoon")) e.Use(middleware.CORSWithConfig(middleware.CORSConfig{ AllowOrigins: []string{"*"}, AllowHeaders: []string{"*"}, @@ -311,13 +316,13 @@ func New(args *Args) (*Server, error) { if err != nil { return nil, fmt.Errorf("failed to connect to postgres: %w", err) } - args.Logger.Info("connected to PostgreSQL database") + logger.Info("connected to PostgreSQL database") default: gdb, err = gorm.Open(sqlite.Open(args.DbName), &gorm.Config{}) if err != nil { return nil, fmt.Errorf("failed to open sqlite database: %w", err) } - args.Logger.Info("connected to SQLite database", "path", args.DbName) + logger.Info("connected to SQLite database", "path", args.DbName) } dbw := db.NewDB(gdb) @@ -360,7 +365,7 @@ func New(args *Args) (*Server, error) { var nonceSecret []byte maybeSecret, err := os.ReadFile("nonce.secret") if err != nil && !os.IsNotExist(err) { - args.Logger.Error("error attempting to read nonce secret", "error", err) + logger.Error("error attempting to read nonce secret", "error", err) } else { nonceSecret = maybeSecret } @@ -398,17 +403,17 @@ func New(args *Args) (*Server, error) { Hostname: args.Hostname, ClientManagerArgs: client.ManagerArgs{ Cli: oauthCli, - Logger: args.Logger, + Logger: args.Logger.With("component", "oauth-client-manager"), }, DpopManagerArgs: dpop.ManagerArgs{ NonceSecret: nonceSecret, NonceRotationInterval: constants.NonceMaxRotationInterval / 3, OnNonceSecretCreated: func(newNonce []byte) { if err := os.WriteFile("nonce.secret", newNonce, 0644); err != nil { - args.Logger.Error("error writing new nonce secret", "error", err) + logger.Error("error writing new nonce secret", "error", err) } }, - Logger: args.Logger, + Logger: args.Logger.With("component", "dpop-manager"), Hostname: args.Hostname, }, }), @@ -535,9 +540,11 @@ func (s *Server) addRoutes() { } func (s *Server) Serve(ctx context.Context) error { + logger := s.logger.With("name", "Serve") + s.addRoutes() - s.logger.Info("migrating...") + logger.Info("migrating...") s.db.AutoMigrate( &models.Actor{}, @@ -554,7 +561,7 @@ func (s *Server) Serve(ctx context.Context) error { &provider.OauthAuthorizationRequest{}, ) - s.logger.Info("starting cocoon") + logger.Info("starting cocoon") go func() { if err := s.httpd.ListenAndServe(); err != nil { @@ -566,7 +573,7 @@ func (s *Server) Serve(ctx context.Context) error { go func() { if err := s.requestCrawl(ctx); err != nil { - s.logger.Error("error requesting crawls", "err", err) + logger.Error("error requesting crawls", "err", err) } }() @@ -584,7 +591,7 @@ func (s *Server) requestCrawl(ctx context.Context) error { logger.Info("requesting crawl with configured relays") - if time.Now().Sub(s.lastRequestCrawl) <= 1*time.Minute { + if time.Since(s.lastRequestCrawl) <= 1*time.Minute { return fmt.Errorf("a crawl request has already been made within the last minute") } @@ -607,18 +614,20 @@ func (s *Server) requestCrawl(ctx context.Context) error { } func (s *Server) doBackup() { + logger := s.logger.With("name", "doBackup") + if s.dbType == "postgres" { - s.logger.Info("skipping S3 backup - PostgreSQL backups should be handled externally (pg_dump, managed database backups, etc.)") + logger.Info("skipping S3 backup - PostgreSQL backups should be handled externally (pg_dump, managed database backups, etc.)") return } start := time.Now() - s.logger.Info("beginning backup to s3...") + logger.Info("beginning backup to s3...") var buf bytes.Buffer if err := func() error { - s.logger.Info("reading database bytes...") + logger.Info("reading database bytes...") s.db.Lock() defer s.db.Unlock() @@ -634,12 +643,12 @@ func (s *Server) doBackup() { return nil }(); err != nil { - s.logger.Error("error backing up database", "error", err) + logger.Error("error backing up database", "error", err) return } if err := func() error { - s.logger.Info("sending to s3...") + logger.Info("sending to s3...") currTime := time.Now().Format("2006-01-02_15-04-05") key := "cocoon-backup-" + currTime + ".db" @@ -669,11 +678,11 @@ func (s *Server) doBackup() { return fmt.Errorf("error uploading file to s3: %w", err) } - s.logger.Info("finished uploading backup to s3", "key", key, "duration", time.Now().Sub(start).Seconds()) + logger.Info("finished uploading backup to s3", "key", key, "duration", time.Now().Sub(start).Seconds()) return nil }(); err != nil { - s.logger.Error("error uploading database backup", "error", err) + logger.Error("error uploading database backup", "error", err) return } @@ -681,27 +690,29 @@ func (s *Server) doBackup() { } func (s *Server) backupRoutine() { + logger := s.logger.With("name", "backupRoutine") + if s.s3Config == nil || !s.s3Config.BackupsEnabled { return } if s.s3Config.Region == "" { - s.logger.Warn("no s3 region configured but backups are enabled. backups will not run.") + logger.Warn("no s3 region configured but backups are enabled. backups will not run.") return } if s.s3Config.Bucket == "" { - s.logger.Warn("no s3 bucket configured but backups are enabled. backups will not run.") + logger.Warn("no s3 bucket configured but backups are enabled. backups will not run.") return } if s.s3Config.AccessKey == "" { - s.logger.Warn("no s3 access key configured but backups are enabled. backups will not run.") + logger.Warn("no s3 access key configured but backups are enabled. backups will not run.") return } if s.s3Config.SecretKey == "" { - s.logger.Warn("no s3 secret key configured but backups are enabled. backups will not run.") + logger.Warn("no s3 secret key configured but backups are enabled. backups will not run.") return } diff --git a/test.go b/test.go index 1cfa14b..84da35e 100644 --- a/test.go +++ b/test.go @@ -32,7 +32,7 @@ func runFirehoseConsumer(relayHost string) error { u.Path = "xrpc/com.atproto.sync.subscribeRepos" conn, _, err := dialer.Dial(u.String(), http.Header{ - "User-Agent": []string{fmt.Sprintf("hot-topic/0.0.0")}, + "User-Agent": []string{"cocoon-test/0.0.0"}, }) if err != nil { return fmt.Errorf("subscribing to firehose failed (dialing): %w", err)